Skip to content

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 验收项)。

动手练习

  1. 实现 search_web mock 工具,与 get_weather、calc 组成三工具 Agent
  2. 设置 max_iterations=2,观察复杂问题如何被截断
  3. 把 stream 事件映射为前端可展示的步骤列表

本节小结

@tool 注册函数为 StructuredTool;create_tool_calling_agent + AgentExecutor 是主流组合;务必设 max_iterations 并用 streamEvents 可观测。