Agent 代码沙箱工具部署

最近更新时间:2025-10-10 10:36:21

我的收藏
本文档介绍在 TKE 上使用 Langchain 构建调用代码沙箱的 Agent。

简介

Agent 代码沙箱是专为 AI Agent 场景设计的新一代运行时基础设施,提供安全隔离的云端执行环境,可防止 Agent 访问或篡改系统外资源。同时,支持多种编程语言,具备快速启动、高并发等特性,能满足多样化的 AI 任务需求。
llm-sandbox 是一个轻量级且便携的沙盒环境,专门设计用于在安全隔离的环境中运行大语言模型(LLM)生成的代码,该开源项目通过 Docker 容器提供了一个易于使用的接口,用于设置、管理和执行代码,简化了运行 LLM 生成代码的过程。
Langchain 是一个用于开发由大型语言模型(LLM)驱动的应用程序的框架,能简化 LLM 应用生命周期各阶段,可借助其开源组件、第三方集成构建有状态智能体。

环境准备

已创建并部署好 TKE 集群,如果您还没有集群,请参见 创建集群
已创建节点池,并且节点池内有至少2个节点,机型推荐 SA5.LARGE8。

环境验证

验证 Python 版本,推荐版本3.11.6。
python --version
预期结果:
Python 3.11.6
验证 pip 版本,推荐版本23.3.1,预期结果:
pip 23.3.1 from /usr/lib/python3.11/site-packages/pip (python 3.11)
如果上述依赖不存在,可以运行以下命令安装依赖。
yum install python3 python3-pip -y #安装python3和pip
安装 llm-sandbox、langchain 和 langchain_openai。
pip install 'llm-sandbox[k8s]' langchain langchain_openai #安装依赖

构建一个工具调用的 Agent

我们将在集群中使用 Langchain 框架构建一个可以调用代码沙箱工具的 Agent,并且可以在沙箱中编写代码并返回结果。

步骤1:定义沙箱工具

代码沙箱提供执行代码的环境,并使用 K8s 作为后端,pod 作为执行代码的隔离环境。
@tool
def run_code(lang: str, code: str, libraries: list | None = None) -> str:
"""Run code in a sandboxed environment.

:param lang: The language of the code, must be one of ['python', 'java', 'javascript', 'cpp', 'go', 'ruby'].
:param code: The code to run.
:param libraries: The libraries to use, it is optional.
:return: The output of the code.
"""
with SandboxSession(lang=lang,backend=SandboxBackend.KUBERNETES,kube_namespace="default", verbose=False) as session:
return session.run(code, libraries).stdout

步骤2:接入模型

场景接入使用 OpenAI 兼容接口的大模型,可以从 Deepseek 等模型服务提供商处获取 API_KEY,或在 TKE 上部署自建大模型。
llm = ChatOpenAI(
model="xxx", # 填入模型名称,例如:deepseek-chat
temperature=0,
max_retries=2,
api_key="xxxx", # 添入账户密钥,例如:sk-4***********************7
base_url="xxxxx",# 添入API端点地址,例如:https://api.deepseek.com
)

步骤3:构建工具调用 Agent

使用 Langchain 构建工具调用 Agent,选择代码沙箱工具为可用的工具。
prompt = hub.pull("hwchase17/openai-functions-agent")
tools = [run_code]
agent = create_tool_calling_agent(llm, tools, prompt) # type: ignore[arg-type]
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) # type: ignore[arg-type]

验证代码沙箱

设计如下四个执行代码的测试用例:
output = agent_executor.invoke({
"input": "Write python code to calculate Pi number by Monte Carlo method then run it."
})
logger.info("Agent: %s", output)

output = agent_executor.invoke({"input": "Write python code to calculate the factorial of a number then run it."})
logger.info("Agent: %s", output)

output = agent_executor.invoke({"input": "Write python code to calculate the Fibonacci sequence then run it."})
logger.info("Agent: %s", output)

output = agent_executor.invoke({"input": "Calculate the sum of the first 10000 numbers."})
logger.info("Agent: %s", output)
预期结果:
完成上述步骤执行,可以在集群中运行代码,查看运行效果,示例的效果如下:

> Entering new AgentExecutor chain...
HTTP Request: POST http://xxxx/v1/chat/completions "HTTP/1.1 200 OK"

Invoking: `run_code` with `{'lang': 'python', 'code': 'import random\\n\\ndef estimate_pi(n):\\n num_point_circle = 0\\n num_point_total = 0\\n \\n for _ in range(n):\\n x = random.uniform(0,1)\\n y = random.uniform(0,1)\\n distance = x**2 + y**2\\n if distance <= 1:\\n num_point_circle += 1\\n num_point_total += 1\\n \\n return 4 * num_point_circle/num_point_total\\n\\nprint(estimate_pi(1000000))'}`


3.142396
HTTP Request: POST http://xxxxx/v1/chat/completions "HTTP/1.1 200 OK"
The calculated value of Pi using the Monte Carlo method with 1,000,000 sample points is approximately **3.142396**.

This is a stochastic method, so the result will vary slightly each time you run it. The accuracy generally improves with a larger number of sample points.

> Finished chain.
Agent: {'input': 'Write python code to calculate Pi number by Monte Carlo method then run it.', 'output': 'The calculated value of Pi using the Monte Carlo method with 1,000,000 sample points is approximately **3.142396**.\\n\\nThis is a stochastic method, so the result will vary slightly each time you run it. The accuracy generally improves with a larger number of sample points.'}


> Entering new AgentExecutor chain...
HTTP Request: POST http://xxxxx/v1/chat/completions "HTTP/1.1 200 OK"

Invoking: `run_code` with `{'lang': 'python', 'code': 'def factorial(n):\\n if n == 0:\\n return 1\\n else:\\n return n * factorial(n-1)\\n\\nnumber = 5\\nprint(f"The factorial of {number} is {factorial(number)}")'}`


The factorial of 5 is 120
HTTP Request: POST http://xxxxx/v1/chat/completions "HTTP/1.1 200 OK"
Here is the Python code to calculate the factorial of a number:

```python
def factorial(n):
if n < 0:
return "Factorial is not defined for negative numbers"
elif n == 0 or n == 1:
return 1
else:
result = 1
for i in range(2, n + 1):
result *= i
return result

number = 5
print(f"The factorial of {number} is {factorial(number)}")
```

And as shown in the output, the factorial of 5 is 120.

> Finished chain.
Agent: {'input': 'Write python code to calculate the factorial of a number then run it.', 'output': 'Here is the Python code to calculate the factorial of a number:\\n\\n```python\\ndef factorial(n):\\n if n < 0:\\n return "Factorial is not defined for negative numbers"\\n elif n == 0 or n == 1:\\n return 1\\n else:\\n result = 1\\n for i in range(2, n + 1):\\n result *= i\\n return result\\n\\nnumber = 5\\nprint(f"The factorial of {number} is {factorial(number)}")\\n```\\n\\nAnd as shown in the output, the factorial of 5 is 120.'}


> Entering new AgentExecutor chain...
HTTP Request: POST http://xxxxx/v1/chat/completions "HTTP/1.1 200 OK"

Invoking: `run_code` with `{'lang': 'python', 'code': '\\ndef fibonacci(n):\\n if n <= 0:\\n return []\\n elif n == 1:\\n return [0]\\n elif n == 2:\\n return [0, 1]\\n else:\\n fib_seq = [0, 1]\\n for i in range(2, n):\\n fib_seq.append(fib_seq[i-1] + fib_seq[i-2])\\n return fib_seq\\n\\n# Calculate the first 10 numbers in the Fibonacci sequence\\nresult = fibonacci(10)\\nprint(result)\\n'}`


[0, 1, 1, 2, 3, 5, 8, 13, 21, 34]
HTTP Request: POST http://xxxx/v1/chat/completions "HTTP/1.1 200 OK"

Invoking: `run_code` with `{'lang': 'python', 'code': 'def fibonacci(n):\\n fib_sequence = [0, 1]\\n while len(fib_sequence) < n:\\n fib_sequence.append(fib_sequence[-1] + fib_sequence[-2])\\n return fib_sequence[:n]\\n\\nprint(fibonacci(10))'}`
responded: The previous attempt to run the code did not produce any output. I will now write the Python code to calculate the Fibonacci sequence and execute it properly.



[0, 1, 1, 2, 3, 5, 8, 13, 21, 34]
HTTP Request: POST http://xxxx/v1/chat/completions "HTTP/1.1 200 OK"
Here is the Python code to calculate the Fibonacci sequence up to the 10th term:

```python
def fibonacci(n):
fib_sequence = [0, 1]
while len(fib_sequence) < n:
next_value = fib_sequence[-1] + fib_sequence[-2]
fib_sequence.append(next_value)
return fib_sequence[:n]

# Calculate the first 10 Fibonacci numbers
result = fibonacci(10)
print(result)
```

When executed, this code produces the following output:
[0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

> Finished chain.
Agent: {'input': 'Write python code to calculate the Fibonacci sequence then run it.', 'output': 'Here is the Python code to calculate the Fibonacci sequence up to the 10th term:\\n\\n```python\\ndef fibonacci(n):\\n fib_sequence = [0, 1]\\n while len(fib_sequence) < n:\\n next_value = fib_sequence[-1] + fib_sequence[-2]\\n fib_sequence.append(next_value)\\n return fib_sequence[:n]\\n\\n# Calculate the first 10 Fibonacci numbers\\nresult = fibonacci(10)\\nprint(result)\\n```\\n\\nWhen executed, this code produces the following output:\\n[0, 1, 1, 2, 3, 5, 8, 13, 21, 34]'}


> Entering new AgentExecutor chain...
HTTP Request: POST http://xxxx/v1/chat/completions "HTTP/1.1 200 OK"

Invoking: `run_code` with `{'lang': 'python', 'code': 'print(sum(range(1, 10001)))'}`


50005000
HTTP Request: POST http://xxxxx/v1/chat/completions "HTTP/1.1 200 OK"
The sum of the first 10,000 numbers is 50,005,000.

> Finished chain.
Agent: {'input': 'Calculate the sum of the first 10000 numbers.', 'output': 'The sum of the first 10,000 numbers is 50,005,000.'}
可参考的完整代码如下:
import logging

from langchain import hub
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

from llm_sandbox import SandboxSession, SandboxBackend

logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger(__name__)


@tool
def run_code(lang: str, code: str, libraries: list | None = None) -> str:
"""Run code in a sandboxed environment.

:param lang: The language of the code, must be one of ['python', 'java', 'javascript', 'cpp', 'go', 'ruby'].
:param code: The code to run.
:param libraries: The libraries to use, it is optional.
:return: The output of the code.
"""
with SandboxSession(lang=lang,backend=SandboxBackend.KUBERNETES,kube_namespace="default", verbose=False) as session:
return session.run(code, libraries).stdout


if __name__ == "__main__":
# llm = ChatOpenAI(model="gpt-4.1-nano", temperature=0)
llm = ChatOpenAI(
model="qwen3-coder",
temperature=0,
max_retries=2,
api_key="1d4a46b238eabef25101b5ff1dc36a150",
base_url="http://49.233.239.193:60000/v1",
)
prompt = hub.pull("hwchase17/openai-functions-agent")
tools = [run_code]

agent = create_tool_calling_agent(llm, tools, prompt) # type: ignore[arg-type]
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) # type: ignore[arg-type]
output = agent_executor.invoke({
"input": "Write python code to calculate Pi number by Monte Carlo method then run it."
})
logger.info("Agent: %s", output)

output = agent_executor.invoke({"input": "Write python code to calculate the factorial of a number then run it."})
logger.info("Agent: %s", output)

output = agent_executor.invoke({"input": "Write python code to calculate the Fibonacci sequence then run it."})
logger.info("Agent: %s", output)

output = agent_executor.invoke({"input": "Calculate the sum of the first 10000 numbers."})
logger.info("Agent: %s", output)

常见问题

pip 安装时无法找到 llm-sandbox 包?

Python 和 pip 版本过低,更新 python 和 pip 的版本,推荐 python3.11 以上。

运行测试代码时,遇到权限报错返回?

开启集群内网访问,并更新 config 文件。