为了提供 Agent 高度的灵活性,有些时候会需要 Agent 生成并执行一些代码,tRPC-Agent 框架为这种场景支持了 CodeExecutor。
当启用了该功能之后,如果LLM返回文本中有代码片段时,框架会调用相应的 CodeExecutor 执行代码,并将执行结果返回给 LLM,LLM 可以根据执行结果继续生成响应。
目前提供下面两种代码执行器:
特点:
- 在 Agent 所在的进程中执行LLM生成的代码
- 非沙箱环境,直接使用本地 Python/Bash 环境,目前仅支持
Python/Bash语言 - 执行速度快,无需 Docker 环境
- 安全警告:LLM 生成的代码可能有风险,不适合用在生产环境
适用场景:
- 开发和测试环境
- 受信任的代码执行场景
- 需要快速迭代和调试的场景
特点:
- Agent派发代码片段到 Docker 容器中执行,目前仅支持
Python/Bash语言 - 沙箱环境,提供更好的隔离和安全性
- 支持自定义 Docker 镜像或 Dockerfile
- 需要 Docker 环境支持
适用场景:
- 生产环境
- 需要执行不可信代码的场景
- 需要环境隔离的场景
创建 LlmAgent 时,构建 CodeExecutor 并配置code_executor参数,即可启用代码执行功能。
from trpc_agent_sdk.agents import LlmAgent
from trpc_agent_sdk.models import LLMModel
from trpc_agent_sdk.models import OpenAIModel
from trpc_agent_sdk.code_executors import BaseCodeExecutor
from trpc_agent_sdk.code_executors import UnsafeLocalCodeExecutor
from trpc_agent_sdk.code_executors import ContainerCodeExecutor
from trpc_agent_sdk.log import logger
def _create_code_executor(code_executor_type: str = "unsafe_local") -> BaseCodeExecutor:
"""Create a code executor.
Args:
code_executor_type: Type of code executor to use. Options:
- "unsafe_local": Use UnsafeLocalCodeExecutor (default, no Docker required)
- "container": Use ContainerCodeExecutor (requires Docker)
- None: Auto-detect from environment variable CODE_EXECUTOR_TYPE,
or default to "unsafe_local"
Returns:
BaseCodeExecutor instance.
Raises:
RuntimeError: If container type is requested but Docker is not available.
The error message will include detailed instructions on how to fix the issue.
"""
# Get executor type from environment variable if not specified
if code_executor_type == "unsafe_local":
return UnsafeLocalCodeExecutor(timeout=10)
elif code_executor_type == "container":
# ContainerCodeExecutor will raise a clear error if Docker is not available
# The error message includes detailed instructions on how to fix the issue
executor = ContainerCodeExecutor(image="python:3-slim", error_retry_attempts=1)
logger.info("ContainerCodeExecutor initialized successfully")
return executor
else:
raise ValueError(f"Invalid code executor type: {code_executor_type}. "
"Valid options are: 'unsafe_local', 'container'")# ...
def create_agent() -> LlmAgent:
"""Create an agent with code execution capabilities.
The agent can:
- Execute Python code blocks generated by the LLM
- Use tools like get_weather_report
- Perform calculations and data processing through code execution
Note: UnsafeLocalCodeExecutor executes code in the current process context.
For production use, consider using ContainerCodeExecutor for better security.
"""
# 选择 unsafe_local
executor = _create_code_executor(code_executor_type="unsafe_local")
agent = LlmAgent(
name="code_assistant",
description="代码执行助手",
model=_create_model(), # You can change this to your preferred model
instruction=INSTRUCTION,
code_executor=executor, # Enables code execution functionality
)
return agent
root_agent = create_agent()执行结果示例:
# ...
def create_agent() -> LlmAgent:
"""Create an agent with code execution capabilities.
The agent can:
- Execute Python code blocks generated by the LLM
- Use tools like get_weather_report
- Perform calculations and data processing through code execution
Note: UnsafeLocalCodeExecutor executes code in the current process context.
For production use, consider using ContainerCodeExecutor for better security.
"""
# 选择 container
executor = _create_code_executor(code_executor_type="container")
agent = LlmAgent(
name="code_assistant",
description="代码执行助手",
model=_create_model(), # You can change this to your preferred model
instruction=INSTRUCTION,
code_executor=executor, # Enables code execution functionality
)
return agent
# 使用前确保Docker已安装并运行
# Linux: sudo systemctl start docker
# Windows/Mac: 启动Docker Desktop执行结果示例:
from trpc_agent_sdk.code_executors import UnsafeLocalCodeExecutor, CodeBlockDelimiter
code_executor = UnsafeLocalCodeExecutor(
# 代码执行失败重试次数,默认是2次
error_retry_attempts=2,
# 代码块分隔符,用于识别LLM返回中的代码块
# 默认支持: ```tool_code\n 和 ```python\n
code_block_delimiters=[
CodeBlockDelimiter(start="```python\n", end="\n```"),
CodeBlockDelimiter(start="```tool_code\n", end="\n```"),
],
# 工作目录,如果为空则使用临时目录
work_dir="",
# 代码执行超时时间(秒)
timeout=10,
# 是否在执行后清理临时文件,默认True
clean_temp_files=True,
)from trpc_agent_sdk.code_executors import ContainerCodeExecutor, CodeBlockDelimiter
code_executor = ContainerCodeExecutor(
# Docker镜像名称(必需,与docker_path二选一)
image="python:3-slim",
# Dockerfile路径(必需,与image二选一)
# docker_path="/path/to/Dockerfile",
# 远程Docker的base_url(可选)
# base_url="tcp://remote-docker-host:2375",
# 代码执行失败重试次数,默认是2次
error_retry_attempts=2,
# 代码块分隔符,默认使用 ```tool_code\n
code_block_delimiters=[
CodeBlockDelimiter(start="```tool_code\n", end="\n```"),
],
)Agent会自动识别LLM返回中的代码块并执行。支持的代码块格式:
```python
print("Hello, World!")
```
```tool_code
result = 15 + 27 * 3
print(result)
```代码执行后,结果会以以下格式返回给LLM:
```tool_output
96
```- Python (
python,py,python3) - Bash (
bash,sh)
- Python (
python,py,python3, 空字符串默认为Python) - Bash (
bash,sh)
- 用户查询 → Agent接收用户查询
- LLM生成响应 → LLM生成包含代码块的响应
- 代码提取 → 框架自动提取代码块(根据
code_block_delimiters) - 代码执行 → CodeExecutor执行代码
- 结果返回 → 执行结果返回给LLM
- 最终响应 → LLM根据执行结果生成最终响应
参考:Pcg123 沙箱环境使用示例(示例待补充)
问题: 使用ContainerCodeExecutor时提示Docker连接失败
解决方案:
- Linux: 确保Docker daemon正在运行
sudo systemctl start docker - Windows/Mac: 启动Docker Desktop应用
- 检查Docker权限:
sudo chmod 666 /var/run/docker.sock或添加用户到docker组 - 验证Docker运行:
docker ps - 如果使用远程Docker,检查
base_url配置
问题: 代码执行时间过长导致超时
解决方案:
# UnsafeLocalCodeExecutor设置超时
code_executor = UnsafeLocalCodeExecutor(timeout=30) # 30秒超时问题: 代码执行失败但没有看到错误信息
解决方案:
- 检查
error_retry_attempts设置,增加重试次数 - 查看日志输出,框架会记录详细的错误信息
- 对于ContainerCodeExecutor,检查容器日志
查看完整示例代码:examples/code_executors/agent/agent.py
- 生产环境:强烈建议使用
ContainerCodeExecutor,提供沙箱隔离 - 代码审查:在生产环境部署前,审查LLM生成的代码
- 资源限制:为 ContainerCodeExecutor 设置适当的资源限制
- 权限控制:限制代码执行器的文件系统访问权限
- 网络隔离:根据需要限制容器的网络访问
- UnsafeLocalCodeExecutor:执行速度快,适合快速迭代
- ContainerCodeExecutor:首次启动需要拉取镜像,后续执行速度较快
- 建议在生产环境使用 ContainerCodeExecutor,开发环境可以使用 UnsafeLocalCodeExecutor



