示例:基于SDK的工具开发与应用实践
本示例介绍基于SDK的工具开发与应用实践,通过LangGraph框架开发Agent并集成hw-agentarts-sdk工具,展示从开发到部署的完整流程。
前提条件
已安装必须的Python包,包括hw-agentarts-sdk、langgraph, langchain-openai等。
基于SDK的工具开发与应用实践
- 下载AgentRuntime SDK包及其它依赖包。
pip install agentarts-sdk pip install langgraph pip install langchain-openai - 导入代码解释器及其它包。
import json import os from typing import TypedDict, Union from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage, AIMessage from langchain_core.tools import tool from hw-agentarts-sdk import HuaweiAgentRunApp from hw-agentarts-sdk.tools import code_session from langgraph.prebuilt import ToolNode app = HuaweiAgentRunApp()
- 创建工具。
代码解释器工具创建需要时间拉起,拉起时间大约为5s。
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代码的工具为例,工具接受模型生成的代码和描述,执行代码并返回代码的执行结果。
@tool def execute_python_tool(code: str, description: str = "") -> str | None: with code_session("your_region", "your_code_interpreter_name") as code_client: response = code_client.invoke( operate_type="execute_code", arguments={ "code": code, "language": "python", "clear_context": False } ) return json.dumps(response["result"]) - 编写系统提示词。
SYSTEM_PROMPT = """你是一个通过代码执行验证所有答案的优秀AI助手 验证原则: 1. 当需要代码、算法或计算来验算时,你需要编写代码来验证它们。 2. 使用 execute_python 来测试数学计算、算法和逻辑。 3. 返回答案之前,使用测试脚本来验证你的理解。 4. 只能通过实际的代码执行展示工作过程。 5. 如果存在不确定的情况,详细说明限制条件并做尽可能的验证 方法: - 如果问题涉及编程的概念,就通过代码实现 - 如果要求你计算,编写程序计算并显示具体代码 - 如果需要实现算法,你还要编写测试用例来进行确认 - 记录验证的过程展示给用户 工具: - execute_python: 执行Python代码并获得输出 响应格式: execute_python_tool工具将返回一个json响应,包括: - content: 内容对象的数组,每个对象包含type和text/data"""
- 开发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.bind_tools(tools) # 定义graph状态 class AgentState(TypedDict): messages: list[Union[HumanMessage, SystemMessage, AIMessage]] 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 last_message.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(): query = "告诉我1到100之间最大的随机质数" # 运行Agent result = agent.invoke({ "messages": [HumanMessage(content=query)] }) print(result["messages"][-1].content) if __name__ == '__main__': app.run()