当前位置：首页 > news >正文

LLM Agent 工具调用框架：从 ReAct 到 Function Calling

news 2026/6/22 10:36:24

1. 引言

大语言模型本身只能生成文本，但通过 Agent 框架，它可以调用工具、搜索互联网、执行代码、操作数据库——成为一个真正的"智能体"。本文将从原理到实现，完整构建一个 LLM Agent 系统。

核心概念：

ReAct：Reasoning + Acting 的循环模式
Function Calling：让 LLM 结构化地调用外部函数
Tool Schema：工具的 JSON Schema 描述
Agent Loop：思考 → 调用工具 → 观察结果 → 继续思考

2. ReAct 模式原理

用户提问 → LLM 思考(Thought) → 选择动作(Action) → 执行工具 → 观察结果(Observation) ↑ | └────────── 循环直到得到最终答案 ────────┘

示例交互：

用户: 北京今天的天气怎么样？ Thought: 用户想知道北京今天的天气，我需要调用天气查询工具。 Action: get_weather(city="北京") Observation: 北京今天晴，气温 15-25°C，北风 3-4 级。 Thought: 我已经获得了天气信息，可以回答用户了。 Answer: 北京今天天气晴朗，气温 15-25°C，北风 3-4 级，适合外出。

3. 工具定义

3.1 工具 Schema 格式

tools=[{"type":"function","function":{"name":"get_weather","description":"查询指定城市的当前天气信息","parameters":{"type":"object","properties":{"city":{"type":"string","description":"城市名称，如'北京'、'上海'"},"unit":{"type":"string","enum":["celsius","fahrenheit"],"description":"温度单位，默认摄氏度"}},"required":["city"]}}},{"type":"function","function":{"name":"search_web","description":"搜索互联网获取最新信息","parameters":{"type":"object","properties":{"query":{"type":"string","description":"搜索关键词"},"num_results":{"type":"integer","description":"返回结果数量","default":5}},"required":["query"]}}},{"type":"function","function":{"name":"execute_python","description":"执行 Python 代码并返回结果","parameters":{"type":"object","properties":{"code":{"type":"string","description":"要执行的 Python 代码"}},"required":["code"]}}}]

3.2 工具实现

importrequestsimportsubprocessimportjsondefget_weather(city:str,unit:str="celsius")->str:"""查询天气"""# 使用 wttr.in 免费 APIresp=requests.get(f"https://wttr.in/{city}?format=j1")data=resp.json()current=data["current_condition"][0]temp=current["temp_C"]ifunit=="celsius"elsecurrent["temp_F"]desc=current["weatherDesc"][0]["value"]wind=current["windspeedKmph"]returnjson.dumps({"city":city,"temperature":f"{temp}°{'C'ifunit=='celsius'else'F'}","description":desc,"wind_speed":f"{wind}km/h",},ensure_ascii=False)defsearch_web(query:str,num_results:int=5)->str:"""搜索网页"""# 使用 DuckDuckGo 搜索fromduckduckgo_searchimportDDGSwithDDGS()asddgs:results=list(ddgs.text(query,max_results=num_results))returnjson.dumps(results,ensure_ascii=False)defexecute_python(code:str)->str:"""安全执行 Python 代码"""importsysfromioimportStringIO old_stdout=sys.stdout sys.stdout=StringIO()try:exec(code,{"__builtins__":__builtins__})output=sys.stdout.getvalue()returnoutputifoutputelse"代码执行成功（无输出）"exceptExceptionase:returnf"执行错误:{str(e)}"finally:sys.stdout=old_stdout# 工具注册表TOOL_REGISTRY={"get_weather":get_weather,"search_web":search_web,"execute_python":execute_python,}

4. Agent 核心实现

fromopenaiimportOpenAIclassLLMAgent:"""LLM Agent 核心引擎"""def__init__(self,model="gpt-4o"):self.client=OpenAI()self.model=model self.tools=tools# 上面定义的工具 schemaself.max_iterations=10defrun(self,user_message:str)->str:"""运行 Agent 循环"""messages=[{"role":"system","content":"你是一个有用的 AI 助手。你可以使用工具来回答问题。""每次使用工具后，观察结果，然后决定是否需要继续使用工具。""如果已经有足够信息，直接回答用户。"},{"role":"user","content":user_message},]foriterationinrange(self.max_iterations):# 调用 LLMresponse=self.client.chat.completions.create(model=self.model,messages=messages,tools=self.tools,tool_choice="auto",)message=response.choices[0].message messages.append(message)# 检查是否有工具调用ifnotmessage.tool_calls:# 没有工具调用，返回最终回答returnmessage.content# 执行所有工具调用fortool_callinmessage.tool_calls:func_name=tool_call.function.name func_args=json.loads(tool_call.function.arguments)print(f" 🔧 调用工具:{func_name}({func_args})")# 执行工具iffunc_nameinTOOL_REGISTRY:result=TOOL_REGISTRY[func_name](**func_args)else:result=f"错误：未知工具{func_name}"print(f" 📋 结果:{result[:200]}...")# 将结果加入消息messages.append({"role":"tool","tool_call_id":tool_call.id,"content":str(result),})return"达到最大迭代次数，未能得出最终答案。"

5. 使用示例

agent=LLMAgent()# 单步工具调用answer=agent.run("北京今天天气怎么样？")print(answer)# 多步推理answer=agent.run("帮我计算一下，如果我现在投资 10 万元，年化收益 8%，""复利计算 10 年后本息合计是多少？")print(answer)# 复杂任务answer=agent.run("搜索一下 2024 年诺贝尔物理学奖得主是谁，然后用 Python ""计算一下他们从出生到获奖分别经历了多少天。")print(answer)

6. 并行工具调用

# OpenAI 支持并行工具调用# LLM 可以在一次响应中返回多个 tool_callsdefrun_parallel(agent,user_message:str)->str:"""支持并行工具调用的 Agent"""messages=[{"role":"system","content":"你可以同时调用多个工具。"},{"role":"user","content":user_message},]response=agent.client.chat.completions.create(model=agent.model,messages=messages,tools=agent.tools,parallel_tool_calls=True,# 启用并行调用)message=response.choices[0].messageifmessage.tool_calls:# 并行执行所有工具importconcurrent.futureswithconcurrent.futures.ThreadPoolExecutor()asexecutor:futures={}fortcinmessage.tool_calls:func=TOOL_REGISTRY[tc.function.name]args=json.loads(tc.function.arguments)futures[tc.id]=executor.submit(func,**args)# 收集结果messages.append(message)fortcinmessage.tool_calls:result=futures[tc.id].result()messages.append({"role":"tool","tool_call_id":tc.id,"content":str(result),})# 第二轮 LLM 调用，整合结果response=agent.client.chat.completions.create(model=agent.model,messages=messages,tools=agent.tools,)returnresponse.choices[0].message.content

7. 安全考虑

# 工具调用的安全最佳实践classSafeAgent(LLMAgent):"""带安全限制的 Agent"""def__init__(self,**kwargs):super().__init__(**kwargs)self.allowed_tools={"get_weather","search_web"}# 白名单self.max_tool_calls_per_turn=5def_validate_tool_call(self,func_name,func_args):"""验证工具调用安全性"""iffunc_namenotinself.allowed_tools:raisePermissionError(f"工具{func_name}不在白名单中")# 代码执行需要额外检查iffunc_name=="execute_python":code=func_args.get("code","")dangerous=["os.system","subprocess","eval(","exec(","import os"]forpatternindangerous:ifpatternincode:raisePermissionError(f"代码包含危险操作:{pattern}")