示例:基于SDK的工具开发与应用实践
本示例介绍基于SDK的工具开发与应用实践,通过langgraph框架开发Agent并集成工具,展示从开发到部署的完整流程。
前提条件
已安装必须的Python包,包括agentarts-sdk、langgraph、 langchain-openai等。
- 管理面(控制面):使用AK/SK认证,通过HUAWEICLOUD_SDK_AK和HUAWEICLOUD_SDK_SK设置环境变量。获取AK/SK请参考如何获取华为云AK/SK。
- 数据面:使用API_KEY认证,通过HUAWEICLOUD_SDK_CODE_INTERPRETER_API_KEY设置环境变量(可选,也可作为客户端参数传入),SDK内部自动处理。获取API_KEY请参考如何获取沙箱工具的API_KEY。
配置访问端点:通过AGENTARTS_CODEINTERPRETER_DATA_ENDPOINT设置环境变量 ,访问端点获取请参考终端节点,获取到终端节点后,在配置使用时,请配置为http://{域名}。
基于SDK的工具开发与应用实践
- 安装agentarts-sdk包及其它依赖包。
pip install agentarts-sdk pip install langgraph pip install langchain-openai
- 导入代码解释器及其它包。
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()
- 创建工具。
代码解释器工具创建需要时间拉起,拉起时间大约为5s。
from agentarts.sdk.tools import CodeInterpreter def create(): client = CodeInterpreter(region="your_region") code_interpreter = client.create_code_interpreter( name="your_code_interpreter_name", api_key_name="your_api_key_name" )
- 开发工具。 此步骤以开发一个执行python代码的工具为例,工具接受模型生成的代码和描述,执行代码并返回代码的执行结果。获取API_KEY请参考如何获取沙箱工具的API_KEY。
@tool def execute_python_tool(code: str, description: str = "") -> str | None: """Execute Python Code in the sandbox""" api_key = "your_actual_api_key" # your api key with code_session("your_region", "your_code_interpreter_name", 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"])
- 编写系统提示词。
SYSTEM_PROMPT = """你是一个AI助手,可以使用Python代码执行工具来解决问题。 可用工具: - execute_python_tool(code: str, description: str): 执行Python代码 使用原则: 1. 仅在需要精确计算或复杂逻辑时使用工具 2. 简单问题直接回答,无需工具验证 3. 工具调用最多1-2次,避免重复验证 4. 获得结果后立即返回答案 """
- 开发Agent并集成工具。
在此步骤中,请填写拥有使用权限的模型(例如:DeepSeek-V3),并补充模型的api_key 和 base_url。您还可以根据需要自定义模型参数,并将自定义工具集成到Agent中。
# 创建Agent llm = ChatOpenAI( model="DeepSeek-V3", 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()至此,已经完成了一个简易的Agent开发。
- 使用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()启动后,可通过在命令行执行 curl --location http://localhost:8080/invocations --data-raw "{}"来获取请求的结果,此处以部署在本地8080端口为例,请根据实际情况填写url信息。