本文以一个“查询服务状态”的工具为例，从原生 LLM API 的消息往返开始，逐步迁移到 LangChain 的 Model、Message、Tool 和 Agent 抽象， 理清每一层的职责边界。

1. 从原生 Tool Calling 开始

上一篇《LLM 应用基础：Prompt、RAG 与 Tool Calling》已经介绍了原生 Tool Calling Loop。它的核心流程可以概括为：

用户问题
  ↓
调用 LLM
  ↓
响应中有 tool_calls？──否──→ 返回最终答案
  │
  是
  ↓
应用程序校验工具名称和参数
  ↓
应用程序执行本地函数
  ↓
把工具结果追加为 tool message
  ↓
再次调用 LLM

LangChain 并没有改变这个协议，只是把字典、请求组装和循环控制包装成更统一的对象和接口。

2. LangChain 的四个核心对象

LangChain 把原生 API 中的几类数据抽象成可以组合的对象：

原生概念 LangChain 对象 主要职责
模型客户端 ChatDeepSeek 等 Chat Model 接收消息并生成 AIMessage
消息字典 SystemMessage、HumanMessage、AIMessage、ToolMessage 表示对话上下文和工具结果
函数定义 @tool 生成的 Tool 描述参数 Schema，并提供统一调用入口
手写循环 create_agent() 管理模型、工具和消息之间的循环

这几种对象的关系可以先记成：

Message → Model → AIMessage(tool_calls)
                         ↓
                      Tool.invoke()
                         ↓
                    ToolMessage
                         ↓
                       Model

因此，Agent 不是一个脱离模型的“智能函数”，而是围绕模型调用建立起来的一套执行循环。

3. Model 与 Message

本文使用 DeepSeek 的 LangChain 集成。根据当前集成文档，ChatDeepSeek 位于 langchain-deepseek 包中，支持 Tool Calling 的示例模型为 deepseek-chat。运行示例前需要安装依赖并设置 DEEPSEEK_API_KEY：

pip install -U langchain langchain-deepseek
export DEEPSEEK_API_KEY="your-api-key"

最小模型调用如下：

from langchain.messages import HumanMessage, SystemMessage
from langchain_deepseek import ChatDeepSeek


model = ChatDeepSeek(
    model="deepseek-chat",
    temperature=0,
)

response = model.invoke([
    SystemMessage(content="你是一个可靠的运维助手。"),
    HumanMessage(content="Redis timeout 可能是什么原因？"),
])

print(type(response).__name__)  # AIMessage
print(response.content)

model.invoke() 接收一条消息或消息列表，返回一个 AIMessage。消息列表就是当前请求的上下文：

消息类型 作用
SystemMessage 规定模型的角色、规则和约束
HumanMessage 表示用户输入
AIMessage 表示模型输出，可能包含 tool_calls
ToolMessage 表示应用执行工具后返回的结果

LangChain 的消息对象和原生 API 的 role、content 字段是一一对应的。工具调用时，AIMessage 会携带结构化的 tool_calls，而 ToolMessage.tool_call_id 用来表明它对应哪一次工具调用。

4. 用 @tool 声明工具

先定义一个不会产生真实副作用的模拟工具：

from langchain.tools import tool


@tool
def get_service_status(service: str) -> str:
    """查询指定服务的当前状态，仅支持 redis 和 mysql。"""
    statuses = {
        "redis": "running，近 5 分钟无重启",
        "mysql": "running，连接数正常",
    }
    return statuses.get(service.lower(), f"暂不支持服务：{service}")

@tool 主要完成三件事：

  1. 读取函数名、参数类型和文档字符串；
  2. 根据这些信息生成模型可以理解的 Tool Schema；
  3. 提供统一的 .invoke() 调用入口。

函数的类型标注和文档字符串会影响模型对工具的理解，因此不应把它们当作装饰。工具描述至少要说明用途、参数含义、返回值和限制。

但 @tool 不会让模型直接执行 Python。它只是在应用程序中注册了一个可被模型请求的能力。

5. bind_tools()：绑定能力，不执行工具

把工具绑定到模型后，模型才知道当前有哪些工具可以请求：

from langchain.messages import HumanMessage


model_with_tools = model.bind_tools([get_service_status])

ai_message = model_with_tools.invoke([
    HumanMessage(content="查询 redis 服务状态。"),
])

print(ai_message.tool_calls)

一个可能的 tool_calls 结果如下：

[
    {
        "name": "get_service_status",
        "args": {"service": "redis"},
        "id": "call_123",
        "type": "tool_call",
    }
]

此时工具还没有执行。ai_message.tool_calls 只是模型提出的结构化请求，应用程序仍然需要：

  • 检查工具名称是否在白名单中；
  • 校验参数类型、范围和业务权限；
  • 判断这个操作是否需要人工确认；
  • 设置超时、调用次数和重试策略。

尤其是“重启服务”“修改订单”“发送邮件”等有副作用的工具，不能因为模型返回了调用请求就自动执行。

6. 手动完成 Tool Calling Loop

先用 LangChain 手动实现一次完整循环，才能理解 create_agent() 的价值：

from langchain.messages import HumanMessage, SystemMessage, ToolMessage


messages = [
    SystemMessage(content="你是可靠的运维助手，需要服务状态时调用工具。"),
    HumanMessage(content="查询 redis 服务状态，并给出一句结论。"),
]

available_tools = {
    get_service_status.name: get_service_status,
}

for step in range(5):
    ai_message = model_with_tools.invoke(messages)
    messages.append(ai_message)

    if not ai_message.tool_calls:
        print(ai_message.content)
        break

    for tool_call in ai_message.tool_calls:
        tool = available_tools.get(tool_call["name"])

        if tool is None:
            messages.append(ToolMessage(
                content="工具不存在或未被允许调用。",
                tool_call_id=tool_call["id"],
            ))
            continue

        try:
            # 传入完整 tool_call 后，LangChain 会生成对应的 ToolMessage。
            tool_message = tool.invoke(tool_call)
        except Exception:
            # 真实项目中应记录内部错误，但不要把敏感异常直接暴露给模型。
            tool_message = ToolMessage(
                content="工具执行失败，请根据失败信息给出说明。",
                tool_call_id=tool_call["id"],
            )

        messages.append(tool_message)
else:
    raise RuntimeError("工具调用超过最大步数")

这段代码中，messages 的变化是：

第一次请求：
[SystemMessage, HumanMessage]

模型请求工具后：
[SystemMessage, HumanMessage, AIMessage(tool_calls)]

工具执行后：
[SystemMessage, HumanMessage, AIMessage(tool_calls), ToolMessage]

下一次请求：
把完整 messages 再传给模型

tool.invoke(tool_call) 会根据调用请求执行对应的函数，并返回与 tool_call["id"] 关联的 ToolMessage。如果手动构造 ToolMessage，必须使用完全相同的 tool_call_id，否则模型无法可靠地把工具结果和请求对应起来。

这就是 Agent 的基本循环：模型提出调用，应用执行工具，工具结果回到消息历史，模型再决定下一步或生成最终回答。

7. create_agent()：把循环交给 Agent

工具和模型准备好后，可以用 create_agent() 管理上面的循环：

from langchain.agents import create_agent


agent = create_agent(
    model=model,
    tools=[get_service_status],
    system_prompt="你是可靠的运维助手，需要服务状态时调用工具。",
)

result = agent.invoke({
    "messages": [
        {"role": "user", "content": "查询 redis 服务状态，并给出一句结论。"},
    ],
})

final_message = result["messages"][-1]
print(final_message.content)

从行为上看，create_agent() 负责管理：

接收用户消息
  ↓
调用模型
  ↓
检查 tool_calls
  ↓
执行已注册工具
  ↓
追加 ToolMessage
  ↓
再次调用模型，直到得到最终回答

与手动实现相比，应用程序不再需要自己维护循环和消息追加逻辑；但工具本身的权限、业务校验、错误分类和风险审批仍然属于应用程序的职责。

可以用下面的表格理解两者的关系：

能力 手动循环 create_agent()
调用模型 应用程序显式调用 Agent 内部管理
检查 tool_calls 应用程序编写判断 Agent 内部管理
执行已注册工具 应用程序编写分发逻辑 Agent 内部管理
保存消息历史 应用程序维护列表 Agent 状态维护
权限和业务规则 应用程序负责 仍需应用程序负责
高风险操作审批 应用程序负责 不会自动获得

因此，create_agent() 更像是一个可靠的流程封装，而不是替业务系统自动拥有“正确决策”的黑盒。

8. Streaming 解决的是响应体验

invoke() 通常等模型生成完整的 AIMessage 后才返回，stream() 则可以逐步得到模型输出：

for chunk in model.stream("用三句话解释什么是 LangChain"):
    print(chunk.text, end="", flush=True)

这里的流是模型文本流。Agent 还可以流出每一步的状态更新，例如：

for update in agent.stream(
    {"messages": [{"role": "user", "content": "查询 redis 服务状态。"}]},
    stream_mode="updates",
):
    print(update)

两者关注点不同：

model.stream()
    └── 主要观察模型生成的文本片段

agent.stream(..., stream_mode="updates")
    └── 观察模型步骤、工具调用和状态更新

聊天窗口只需要逐字展示答案时，模型文本流通常已经足够；需要展示“正在查询服务状态”等过程时，再使用 Agent 的状态流。参考 LangChain Streaming。

9. LangChain 替我们做了什么

LangChain 的核心价值可以概括为三种统一：

统一模型调用：ChatModel.invoke / stream / batch
统一消息对象：HumanMessage / AIMessage / ToolMessage
统一工具描述：@tool / Tool Schema / tool.invoke

这些统一接口让模型、工具和 Agent 更容易组合，也让应用程序可以在保留控制点的同时逐步提高抽象层级。

但 LangChain 不会替我们保证：

  • 模型一定理解了业务问题；
  • 工具参数一定符合业务规则；
  • 工具一定有权执行副作用；
  • 失败后一定应该重试；
  • 重试不会造成重复写入；
  • 工具结果和最终答案一定准确。

Structured Output 也属于类似的边界：它约束的是模型输出结构，而 Tool Calling 表达的是应用程序可以执行的动作。结构合法不代表业务内容正确，工具调用成功也不代表最终结论正确。

官方参考