LangChain工具与代理

工具（Tool）让模型有能力调用外部函数，比如搜索网络、查询数据库、执行代码。代理（Agent）则是能够自主决定调用哪些工具、按什么顺序执行来完成目标的系统。

1. 什么是工具

工具本质上是一个带有名称和描述的函数。模型会根据描述判断什么时候该调用哪个工具。

用装饰器定义工具（最简单）：

from langchain_core.tools import tool

@tool
def add(a: int, b: int) -> int:
    """将两个整数相加并返回结果。"""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """将两个整数相乘并返回结果。"""
    return a * b

print(add.name)         # add
print(add.description)  # 将两个整数相加并返回结果。
print(add.invoke({"a": 3, "b": 4}))  # 7

函数的文档字符串（docstring）非常重要，模型靠它来理解工具的用途，写得越清楚越好。

2. 将工具绑定到模型

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

load_dotenv()

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气情况。"""
    # 实际项目中这里会调用天气 API
    weather_data = {
        "北京": "晴天，气温 25°C",
        "上海": "多云，气温 22°C",
        "广州": "小雨，气温 28°C",
    }
    return weather_data.get(city, f"暂无 {city} 的天气数据")

@tool
def get_time(city: str) -> str:
    """查询指定城市的当前时间。"""
    import datetime
    return f"{city} 当前时间：{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"

llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools([get_weather, get_time])

response = llm_with_tools.invoke("北京今天天气怎么样？")
print(response)

当模型决定调用工具时，response 中的 tool_calls 字段会包含工具名称和参数，但它不会自动执行工具。

3. 手动处理工具调用

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

load_dotenv()

@tool
def calculate(expression: str) -> str:
    """计算数学表达式，如 '2 + 3 * 4'。只支持基础四则运算。"""
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return str(result)
    except Exception as e:
        return f"计算错误：{e}"

tools = [calculate]
tool_map = {t.name: t for t in tools}

llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools)

messages = [HumanMessage(content="25 乘以 16 等于多少？")]

# 第一轮：模型决定调用工具
response = llm_with_tools.invoke(messages)
messages.append(response)

# 如果模型有工具调用请求
if response.tool_calls:
    for tool_call in response.tool_calls:
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]

        # 执行工具
        result = tool_map[tool_name].invoke(tool_args)

        # 将工具结果添加到消息中
        messages.append(ToolMessage(
            content=str(result),
            tool_call_id=tool_call["id"],
        ))

# 第二轮：模型根据工具结果生成最终回答
final_response = llm_with_tools.invoke(messages)
print(final_response.content)  # 25 乘以 16 等于 400

4. 使用内置代理（ReAct）

手动处理工具调用很繁琐，LangGraph 提供了 create_react_agent 来自动处理这个循环：

pip install langgraph

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

load_dotenv()

@tool
def search_web(query: str) -> str:
    """搜索互联网获取信息。输入搜索关键词，返回相关结果。"""
    # 实际使用时替换为真实搜索 API（如 Tavily、SerpAPI）
    return f"关于 '{query}' 的搜索结果：这是一个模拟结果。"

@tool
def get_word_count(text: str) -> int:
    """统计文本的字数（中英文都支持）。"""
    return len(text)

llm = ChatOpenAI(model="gpt-4o-mini")
agent = create_react_agent(llm, tools=[search_web, get_word_count])

result = agent.invoke({
    "messages": [("human", "帮我搜索一下 LangChain 的最新动态，并统计搜索结果的字数")]
})

# 打印最终回答
print(result["messages"][-1].content)

代理会自动决定：先调用 search_web，得到结果后再调用 get_word_count，最后汇总回答。

5. 常用内置工具

LangChain 提供了很多开箱即用的工具：

Tavily 搜索（推荐）：

pip install tavily-python

from langchain_community.tools.tavily_search import TavilySearchResults
import os

os.environ["TAVILY_API_KEY"] = "your-tavily-key"

search_tool = TavilySearchResults(max_results=3)
results = search_tool.invoke("LangChain 最新版本")

Python REPL（执行 Python 代码）：

from langchain_experimental.tools import PythonREPLTool

python_tool = PythonREPLTool()
result = python_tool.invoke("print([i**2 for i in range(10)])")
print(result)

注意

PythonREPLTool 会执行真实的 Python 代码，存在安全风险，生产环境需要在沙箱中运行。

6. 自定义工具（带校验）

对于复杂工具，可以用 BaseTool 和 Pydantic 来定义输入结构：

from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type

class DatabaseQueryInput(BaseModel):
    table: str = Field(description="要查询的数据表名称")
    limit: int = Field(description="返回的最大记录数", default=10)

class DatabaseQueryTool(BaseTool):
    name: str = "database_query"
    description: str = "查询数据库中的记录，返回 JSON 格式的结果"
    args_schema: Type[BaseModel] = DatabaseQueryInput

    def _run(self, table: str, limit: int = 10) -> str:
        # 实际项目中这里执行真实的数据库查询
        return f"从 {table} 表返回 {limit} 条记录：[模拟数据]"

db_tool = DatabaseQueryTool()
result = db_tool.invoke({"table": "users", "limit": 5})
print(result)

7. 代理与对话历史结合

create_react_agent 支持传入对话历史，实现有记忆的代理：

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver

load_dotenv()

@tool
def get_date() -> str:
    """获取今天的日期。"""
    import datetime
    return datetime.date.today().strftime("%Y年%m月%d日")

llm = ChatOpenAI(model="gpt-4o-mini")
memory = MemorySaver()  # 内存中的检查点

agent = create_react_agent(llm, tools=[get_date], checkpointer=memory)

config = {"configurable": {"thread_id": "user_1"}}

# 第一轮
result = agent.invoke(
    {"messages": [HumanMessage(content="今天是什么日期？")]},
    config=config
)
print(result["messages"][-1].content)

# 第二轮（代理记得上一轮对话）
result = agent.invoke(
    {"messages": [HumanMessage(content="所以今年还剩多少天？")]},
    config=config
)
print(result["messages"][-1].content)

总结

工具是带有名称和描述的函数，@tool 装饰器是最简单的定义方式
llm.bind_tools() 将工具能力绑定到模型
模型决定调用工具，但需要代码手动执行工具并将结果回传
create_react_agent 自动处理工具调用循环（ReAct 模式）
MemorySaver 为代理添加对话记忆