j9九游会登录/ 智果(agentarts)智能体平台/ 高代码开发/ / / 示例:基于sdk的工具开发与应用实践
更新时间:2026-07-06 gmt 08:00

示例:基于sdk的工具开发与应用实践-j9九游会登录

本示例介绍如何完全通过编写代码脚本,一次性自动完成沙箱的创建与agent的开发。

示例中分为两个阶段,先通过agentarts sdk创建沙箱,再创建agent并调用沙箱完成任务。

基于sdk的工具开发与应用实践

一、使用agentarts sdk创建沙箱。

  1. 安装python:请确保python 3.10及以上版本已安装。

    大多数linux发行版(如ubuntu)都预装了python,您可以先通过python3 --version检查。如未安装,可以使用如下命令安装:

    sudo apt update
    sudo apt install python3
  2. 执行以下命令创建虚拟环境(建议在python虚拟环境中安装,以避免与系统包产生冲突)。
    # 创建并激活虚拟环境 (linux) 
    python3 -m venv venv 
    source venv/bin/activate
  3. 在您的linux服务器的终端中,运行以下命令安装必要的库,执行以下命令安装agentarts-sdk,langgraph,langchain_openai包。
    # 安装agentarts-sdk包
    pip install agentarts-sdk
    # 安装langgraph包
    pip install langgraph
    # 安装langchain_openai包
    pip install langchain-openai
  4. 配置环境变量。
    # === 用于通过代码创建沙箱 ===
    export huaweicloud_sdk_ak="你的华为云ak"
    export huaweicloud_sdk_sk="你的华为云sk"
    export agentarts_control_endpoint="https://agentarts.cn-southwest-2.myhuaweicloud.com"
  5. 新建一个python文件,用于创建沙箱(例如sandbox.py)并填写以下代码。
    导入代码解释器及其它包。
    import json
    import os
    from typing import annotated, typeddict
    from langchain_core.messages import aimessage, basemessage, humanmessage, systemmessage
    from langchain_core.tools import tool
    from langchain_openai import chatopenai
    from langgraph.graph import end, stategraph
    from langgraph.graph.message import add_messages
    from langgraph.prebuilt import toolnode
    from agentarts.sdk import agentartsruntimeapp
    from agentarts.sdk.tools import code_session
    app = agentartsruntimeapp()
  6. 继续填写如下代码,用于创建沙箱工具(创建的沙箱入站身份方式为api key认证)。
    from agentarts.sdk.tools import codeinterpreter
    def create():
        print("开始创建沙箱,请稍等...")
        client = codeinterpreter(region="cn-southwest-2", auth_type="api_key")  
        code_interpreter = client.create_code_interpreter(
            name="填写一个沙箱名称",
            api_key_name="填写一个api key的名称"
        )
        print("沙箱创建成功!快去控制台刷新看看吧。")
        # 打印出创建结果的详细信息
        print(code_interpreter)
    if __name__ == '__main__':
        create()    

    代码中的沙箱名称、api key的名称需要用户自行命名,代码中的命名操作等同于在agentarts平台中的手动创建操作。

    • 沙箱名称:以小写字母开头,小写字母或数字结尾,中间可包含数字、小写字母、中划线。字符数:24/40。
    • api key名称:只能包含如下字符:英文字母、数字或特殊字符(_-)。
    图1 在agentarts平台创建沙箱
  7. 执行python sandbox.py创建沙箱。

    代码解释器沙箱工具创建需要时间拉起,拉起时间大约为5s。创建完成后,可以在agentarts平台看到创建好的沙箱工具。

    图2 查看创建的沙箱工具

二、创建agent并测试沙箱

  1. 代码解释器创建完成后,在列表可查看已创建的代码解释器的名称。记录该名称,并获取该沙箱工具的域名、api key。

    沙箱工具的域名、api key获取方法如下:

    1. 单击沙箱工具名称。
    2. 进入“配置信息”页面后,获取域名。
      图3 获取沙箱域名
    3. 单击“urn”链接,在弹出的窗口中获取沙箱的api key。
      图4 urn链接
      图5 获取沙箱api key
  2. 配置环境变量。
    export agentarts_codeinterpreter_data_endpoint="http://您的沙箱域名"
    export model_api_key="你的大模型api密钥"
    export base_url="https://api.modelarts-maas.com/openai/v1"

    其中,agentarts_codeinterpreter_data_endpoint请配置为http://{沙箱域名},获取方法请参考前面的步骤。

    由于示例中使用华为云maas服务中的模型,因此需要创建api key。

    https://api.modelarts-maas.com/openai/v1表示使用openai兼容接口。

  3. 新建一个python代码文件(例如app.py)将以下步骤的代码脚本按照顺序组合进去。
    导入代码解释器及其它包。
    import json
    import os
    from typing import annotated, typeddict
    from langchain_core.messages import aimessage, basemessage, humanmessage, systemmessage
    from langchain_core.tools import tool
    from langchain_openai import chatopenai
    from langgraph.graph import end, stategraph
    from langgraph.graph.message import add_messages
    from langgraph.prebuilt import toolnode
    from agentarts.sdk import agentartsruntimeapp
    from agentarts.sdk.tools import code_session
    app = agentartsruntimeapp()
  4. 代码文件中填写如下脚本。用于将“连接沙箱并执行代码”这一操作包装成一个标准的函数工具。当agent中的大模型发现需要执行代码时,会将写好的代码传递给这个工具,工具帮把代码发送到沙箱运行,再把结果拿回来给大模型。
    @tool
    def execute_python_tool(code: str, description: str = "") -> str | none:
        """execute python code in the sandbox"""
        api_key = "替换为获取的沙箱api key" # your api key
        with code_session("cn-southwest-2", "替换为创建的沙箱名称", api_key=api_key) as code_client:
            response = code_client.invoke(
                operate_type="execute_code",
                api_key=api_key,
                arguments={
                    "code": code,
                    "language": "python",
                    "clear_context": false
                }
            )
        return json.dumps(response["result"])
    
  5. 代码文件中添加如下代码,用于编写系统提示词。
    system_prompt = """你是一个ai助手,可以使用python代码执行工具来解决问题。
    可用工具:
    - execute_python_tool(code: str, description: str): 执行python代码
    使用原则:
    1. 仅在需要精确计算或复杂逻辑时使用工具
    2. 简单问题直接回答,无需工具验证
    3. 工具调用最多1-2次,避免重复验证
    4. 获得结果后立即返回答案
    """
  6. 代码文件中添加如下代码,用于定义一个agent并集成沙箱。

    示例中使用deepseek-v4-flash模型,如果需要更换其他模型,请,获取模型接口的model参数值(注意代码使用了openai格式的调用方法,需获取对应格式的model参数值)。

    图6 获取maas服务model参数值
    # 创建agent
    llm = chatopenai(
        model="deepseek-v4-flash",
        api_key=os.environ.get("model_api_key", ""),
        base_url=os.environ.get("base_url", ""),
        max_tokens=1000,
        temperature=0.7,
    )
    # 创建工具列表
    tools = [execute_python_tool]
    # 工具绑定agent
    llm = llm.bind_tools(tools)
    # 定义graph状态
    class agentstate(typeddict):
        messages: annotated[list[basemessage], add_messages]
    def call_model(state: agentstate):
        """调用模型并返回响应"""
        if not state["messages"] or all(
            not isinstance(msg, systemmessage) for msg in state["messages"]
        ):
            messages = [systemmessage(content=system_prompt)]   state["messages"]
        else:
            messages = state["messages"]
        response = llm.invoke(messages)
        return {"messages": [response]}
    def should_continue(state):
        """判断是否继续使用工具"""
        last_message = state["messages"][-1]
        if isinstance(last_message, aimessage):
            has_tool_calls = bool(last_message.tool_calls)
            if has_tool_calls:
                return "tools"
        return end
    # 创建langgraph工作流
    workflow = stategraph(agentstate)
    workflow.add_node("agent", call_model)
    workflow.add_node("tools", toolnode(tools))
    # 设置入口
    workflow.set_entry_point("agent")
    # 添加边
    workflow.add_conditional_edges("agent", should_continue, {"tools": "tools", "__end__": "__end__"})
    workflow.add_edge("tools", "agent")
    agent = workflow.compile()
  7. 拼接以下代码,用于使用agent启动一个本地服务器,并提供一个对外接收提问的接口。
    @app.entrypoint
    def agent_chat(payload: dict):
        query = "告诉我1到100之间最大的质数"
        # 运行agent
        result = agent.invoke({
            "messages": [humanmessage(content=query)]
        })
        return result["messages"][-1].content
    if __name__ == '__main__':
        app.run()
  8. 在服务器中执行python app.py命令运行示例代码(如果在创建python代码文件时使用了其他名称,请按实际名称执行)。
  9. 打开一个新的终端窗口(保持原窗口运行),使用curl命令进行测试。测试完成后,可以使用ctrl c停止运行的进程。
    上述步骤中已经在代码里面内置了问题“告诉我1到100之间最大的质数”,agent会将该计算过程放到沙箱中运行,并返回结果。
    curl --location http://localhost:8080/invocations --data-raw "{}"

相关文档

网站地图