容器服务 Agent 代码沙箱工具部署

这篇档介绍在 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 作为执行代码的隔离环境。

@tooldef 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.142396HTTP 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 120HTTP Request: POST http://xxxxx/v1/chat/completions "HTTP/1.1 200 OK"Here is the Python code to calculate the factorial of a number:“`pythondef 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 resultnumber = 5print(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:“`pythondef 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 numbersresult = 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)))'}`50005000HTTP 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 loggingfrom langchain import hubfrom langchain.agents import AgentExecutor, create_tool_calling_agentfrom langchain_core.tools import toolfrom langchain_openai import ChatOpenAIfrom llm_sandbox import SandboxSession, SandboxBackendlogging.basicConfig(level=logging.INFO, format="%(message)s")logger = logging.getLogger(__name__)@tooldef 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).stdoutif __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 文件。

赞 (0)
上一篇 4天前
下一篇 4天前