AI

LangChain

使用 LangChain 开发支持模型调用、工具、结构化输出、记忆和 RAG 的 AI 应用。

LangChain 是一个用于开发 LLM 应用和 AI Agent 的开源框架。它统一了不同模型提供商的调用方式，并提供 Messages、Tools、Agents、Middleware、Retrieval 和 Structured Output 等组件。

LangChain 的重点不是封装一次模型请求，而是把模型与外部数据、工具和应用控制逻辑组合起来。开发者可以先使用高层 API 快速构建 Agent，再在需要精确控制执行路径时下沉到 LangGraph。

LangChain 生态

LangChain 生态中的几个项目各有职责：

项目主要职责
LangChain模型、Tool、Agent、Middleware 和 Retrieval 等开发框架
LangGraph有状态、可恢复的底层 Agent 编排运行时
LangSmithTrace、评估、测试和生产可观测性
Deep Agents在 LangGraph 之上提供规划、子 Agent、文件系统和上下文管理能力

create_agent 创建的 Agent 底层运行在 LangGraph 上，因此也支持 State、Checkpoint、Streaming 和 Human-in-the-loop。大多数项目可以从 LangChain 开始，不需要一开始就手写 Graph。

核心组件

Models

LangChain 使用统一接口连接不同模型提供商。应用可以使用 invoke()、stream()、batch() 和对应的异步方法调用模型，而不需要让业务逻辑依赖某一个 SDK。

from langchain.chat_models import init_chat_model


model = init_chat_model("openai:gpt-5.4")

response = model.invoke(
    [
        {"role": "system", "content": "你是一名简洁的技术助手。"},
        {"role": "user", "content": "用一句话解释 Tool Calling。"},
    ]
)

print(response.content)

模型集成通常放在独立的 Provider Package 中。更换模型时除了修改模型名称，还需要安装相应集成包并配置该提供商的凭据。

Messages

Messages 是模型交互的标准数据结构。常见角色包括：

  • System Message：定义模型角色和约束；
  • Human Message：用户输入；
  • AI Message：模型输出，可能包含 Tool Call；
  • Tool Message：工具执行结果。

使用 Messages 而不是拼接字符串，可以保留角色、Tool Call、Token Usage 和其他元数据。

Tools

Tool 让模型可以读取实时数据或执行外部操作。普通 Python 函数可以通过 @tool 转换为 Tool：

from langchain.tools import tool


@tool
def get_weather(city: str) -> str:
    """查询指定城市当前的天气。"""
    weather = {
        "北京": "晴，26°C",
        "上海": "多云，29°C",
        "深圳": "阵雨，31°C",
    }
    return weather.get(city, f"暂时没有 {city} 的天气数据")

函数名、类型标注和 Docstring 会共同构成 Tool Schema。描述越准确，模型越容易在正确的时机选择正确的工具。

Tool 应保持单一职责。查询和修改操作最好拆开，敏感参数应由运行时注入，不要暴露给模型。

开发一个 Agent

安装依赖

下面以 OpenAI Provider 为例：

python -m venv .venv
source .venv/bin/activate
pip install -U "langchain[openai]"

通过环境变量配置凭据，不要把 API Key 写进源代码：

export OPENAI_API_KEY="..."

创建 Tool-calling Agent

from langchain.agents import create_agent
from langchain.tools import tool


@tool
def get_weather(city: str) -> str:
    """查询指定城市当前的天气。"""
    weather = {
        "北京": "晴，26°C",
        "上海": "多云，29°C",
        "深圳": "阵雨，31°C",
    }
    return weather.get(city, f"暂时没有 {city} 的天气数据")


agent = create_agent(
    model="openai:gpt-5.4",
    tools=[get_weather],
    system_prompt="你是天气助手。需要天气数据时调用工具，不要编造结果。",
)

result = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "上海和深圳哪个城市更热？"}
        ]
    }
)

result["messages"][-1].pretty_print()

Agent 的执行循环是：

用户输入 → 模型判断 → 调用 Tool → 读取结果 → 模型继续判断 → 最终回答

如果模型一次生成多个 Tool Call，Agent 可以执行这些工具后再把结果交回模型。当模型不再请求工具时，循环结束。

Structured Output

如果下游代码需要稳定的数据结构，不应该依赖自然语言解析。可以通过 Pydantic Schema 要求 Agent 返回结构化结果：

from pydantic import BaseModel, Field
from langchain.agents import create_agent


class WeatherResult(BaseModel):
    city: str = Field(description="城市名称")
    condition: str = Field(description="天气情况")
    temperature_celsius: float = Field(description="摄氏温度")


structured_agent = create_agent(
    model="openai:gpt-5.4",
    tools=[get_weather],
    response_format=WeatherResult,
)

result = structured_agent.invoke(
    {"messages": [{"role": "user", "content": "上海天气如何？"}]}
)

weather = result["structured_response"]
print(weather.temperature_celsius)

对于支持原生 Structured Output 的模型，LangChain 会优先使用 Provider 能力；否则可以使用 Tool Calling 生成符合 Schema 的结果。

Memory

Agent 的短期记忆保存在 State 中。加入 Checkpointer 后，可以使用 thread_id 隔离和恢复不同会话：

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver


checkpointer = InMemorySaver()

agent = create_agent(
    model="openai:gpt-5.4",
    tools=[get_weather],
    checkpointer=checkpointer,
)

config = {"configurable": {"thread_id": "demo-thread"}}

agent.invoke(
    {"messages": [{"role": "user", "content": "我准备去上海。"}]},
    config,
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "那里的天气如何？"}]},
    config,
)

同一个 thread_id 会继续之前的消息历史，不同 Thread 之间相互隔离。

InMemorySaver 只适合开发和测试。生产环境应使用持久化 Checkpointer，并对长对话进行裁剪或摘要，避免上下文无限增长。

Middleware

Middleware 可以在 Agent 循环的关键位置插入控制逻辑，而不需要修改 Agent 本身。常见用途包括：

  • 自动摘要过长的对话；
  • 限制模型或 Tool 调用次数；
  • 重试和模型降级；
  • PII 检测与脱敏；
  • 动态选择模型和 Tool；
  • 记录日志、延迟和 Token 使用量；
  • 在敏感操作前请求人工审批。

Human-in-the-loop

下面的 Agent 会在执行订单操作前暂停，等待人工批准、修改或拒绝：

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command


@tool
def submit_order(item: str, quantity: int) -> str:
    """提交商品订单。"""
    return f"已提交 {quantity} 件 {item}"


agent = create_agent(
    model="openai:gpt-5.4",
    tools=[submit_order],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={"submit_order": True},
        )
    ],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "approval-thread"}}

agent.invoke(
    {"messages": [{"role": "user", "content": "订购两件显示器"}]},
    config,
)

agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config,
)

生产环境中的写操作必须具备幂等性，并使用持久化 Checkpointer，确保审批等待期间进程重启也不会丢失 State。

Retrieval 与 RAG

LLM 的上下文长度有限，训练数据也不是实时知识。Retrieval 会在请求时获取相关外部内容，再让模型基于这些内容回答，这就是 RAG 的基础。

LangChain 将 Retrieval 拆成可替换的组件：

Document Loader
      ↓
Text Splitter
      ↓
Embedding Model
      ↓
Vector Store
      ↓
Retriever
      ↓
Prompt + Model

不同场景可以选择不同的 RAG 架构：

架构特点适用场景
2-step RAG每次先检索再生成，流程固定、延迟可预测FAQ、文档问答
Agentic RAGAgent 自己决定何时检索以及使用哪个数据源研究助手、多数据源查询
Hybrid RAG加入查询改写、检索验证和答案检查对质量和可控性要求较高的系统

如果已经有搜索服务、SQL 数据库或知识库，不需要为了使用 LangChain 重新建立向量数据库。可以把现有系统包装成 Retriever 或 Tool。

LangChain 与 LangGraph 的选择

优先使用 LangChain 的场景：

  • 调用不同 Provider 的模型；
  • 开发标准 Tool-calling Agent；
  • 使用 Middleware、Structured Output 或 RAG；
  • 希望快速组合现有集成。

直接使用 LangGraph 的场景：

  • 工作流包含复杂条件分支或并行节点；
  • 需要精确控制 State 和执行路径；
  • 需要暂停、恢复、长时间运行或故障恢复；
  • 需要多个 Agent 或子图协作。

两者不是互斥关系。可以先用 create_agent 构建一个 Agent，再把它作为 Node 或 Subgraph 放入更大的 LangGraph 工作流中。

生产实践

明确 Tool 边界

Tool 的输入必须严格校验。读操作和写操作分离，高风险 Tool 需要权限检查和人工审批。不要让模型直接拼接 SQL、Shell 或其他可执行内容后无条件运行。

使用 Structured Output

模型输出需要进入数据库、API 或后续工作流时，应使用 Schema 校验，而不是依赖正则表达式解析自然语言。

限制 Agent 循环

限制模型调用次数、Tool 调用次数、总超时和 Token 预算。Agent 无法完成任务时应该明确退出，而不是无限重试。

管理上下文

只向模型提供当前任务需要的信息。长对话应裁剪或摘要，大型文档应通过 Retrieval 按需加载。

增加 Trace 与评估

记录 Prompt、模型输出、Tool Call、延迟、错误和执行路径。除了单元测试，还应建立包含正常请求、边界条件和恶意输入的数据集，对 Agent 进行回归评估。

保持 Provider 可替换

把模型初始化、参数和凭据放在配置层。切换 Provider 时仍需要重新评估 Tool Calling、Structured Output、上下文长度、延迟和成本，不能只修改模型名称就认为行为完全一致。

参考资料

© 2013 - 2026 Morgoth
Designed by Canux