Tools 与 Agent
🎯 学习目标
- 用 @tool 装饰器定义带 schema 的工具函数
- 分别构建 ReAct Agent 与 tool_calling Agent
- 使用 stream / astream_events 观察工具调用过程
引言
LangChain Agent 层把 LLM 与 Tools 包进循环;你仍需在 Harness 层做权限、超时与迭代上限(见 5.3、7.5)。
章节正文
第 1 步:定义 Tools
python
from langchain_core.tools import tool
@tool
def get_weather(city: str) -> str:
"""查询城市天气。输入城市中文名。"""
# 实际项目调用天气 API
return f"{city}:晴,28°C"
@tool
def calc(expression: str) -> str:
"""计算数学表达式,如 2*(3+4)"""
return str(eval(expression)) # 生产环境用 safe eval
tools = [get_weather, calc]docstring 会进入 tool schema,影响模型选工具时的理解。
第 2 步:Tool Calling Agent(推荐)
python
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system", "你是助手,可调用工具。用中文回答。"),
("human", "{input}"),
MessagesPlaceholder("agent_scratchpad"),
])
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=5)
executor.invoke({"input": "北京天气怎样,再算 12*8"})max_iterations 是防止死循环的硬上限。
第 3 步:streamEvents 观察链路
python
for event in executor.stream({"input": "上海天气"}):
print(event)
# 或底层 chain
async for ev in executor.astream_events({"input": "..."}, version="v2"):
kind = ev["event"]
if kind == "on_tool_start":
print("工具:", ev["name"], ev["data"].get("input"))
elif kind == "on_tool_end":
print("结果:", ev["data"].get("output"))前端可据此展示「正在查询天气…」等中间状态(阶段二 11.2 验收项)。
动手练习
- 实现 search_web mock 工具,与 get_weather、calc 组成三工具 Agent
- 设置 max_iterations=2,观察复杂问题如何被截断
- 把 stream 事件映射为前端可展示的步骤列表
本节小结
@tool 注册函数为 StructuredTool;create_tool_calling_agent + AgentExecutor 是主流组合;务必设 max_iterations 并用 streamEvents 可观测。