为什么复杂AI项目要用LangChain？

AI训练营7期，1月下旬开班，欢迎咨询

在我们上课过程中有个很常见的问题：做AI项目应该选什么框架？这里最常提及的名词是：Coze、Dify、FastGPT、n8n，然后就是LangChain。

对于我们这批“科班”出身的人来说，最喜欢做的事情是自己手撸代码，因为可控性会更高，除非demo需要几乎不会用Coze这类拖拽低代码平台，如果非要让我选择，LangChain可能是最优解。

LangChain 和 LangGraph 作为当前最常见的 AI Agent 开发框架，他为开发者提供了从组件封装到流程编排的工具链。

随着 LangChain 1.x 与 LangGraph 1.x 的逐步完善，整个体系的生态分工与工程化实践也变得更清晰。

今天就让我们来全面解析这两个框架的核心概念、发展历程、主要功能及实际应用。

LangChain的发展历程

LangChain由Harrison Chase于2022年10月创立，最初作为一个专注于用LLM构建应用"的开源框架。

那时候，ChatGPT刚刚引发全球AI热潮，开发者们急需一种能够快速将LLM与外部数据源、工具、API连接起来的解决方案。

一、占领身位

LangChain的诞生恰逢其时，他其实就是一套最佳实践被抽象出来了，最终提供了一系列关键的抽象层：

1. Models（模型）：对不同大语言模型进行统一封装，提供标准化的推理、对话接口，作为整个系统的基础能力层。
2. Chains（链）：将多个组件串联形成工作流
3. Tools（工具）：为大模型提供外部能力接口
4. Agents（代理）：实现自主决策的工具调用机制
5. Memory（记忆）：负责管理与注入对话历史和中间状态，使大模型在多轮交互中具备上下文连续性。

框架极大地简化了LLM应用的开发流程，使得开发者可以快速搭建问答系统、摘要系统、对话机器人，其实上述的很多功能在初期并没有太有用，初期可以吧LangChain等同于RAG就可以了。

只不过，LangChain 的早期架构采用了单体（Monolithic）设计，所有组件紧密耦合在一起，虽然方便了快速集成，但也带来了扩展性难题，所以我们自己更多是参考他的架构，实际工作起来都是自己玩：

二、快速迭代

如前所述，初期ChatGPT能力本身就很弱，算力也不足，所以在生产环境其实我们是不太用LangChain的。但从23年开始模型的能力几乎每半年变个样，相应着LangChain也在不停迭代，补足自身补足，比如：

1. 模块化设计更合理，各组件可以灵活组合
2. 支持多种模型提供商（OpenAI、Anthropic、Google等）
3. 生态系统丰富，社区贡献了大量集成

只不过这个时期，所有的AI开源框架都会面临相同的问题：

1. API频繁变更，破坏性更新较多
2. 代理逻辑分散，难以维护复杂流程
3. 状态管理能力较弱
4. 对于需要长期记忆、状态持久化的应用支持不足(这里不完全是框架的问题，模型本身能力就很弱)

值得提一嘴，现在大火的Agent在2023年还很弱，当时AgentExecutor 是实现智能体的核心。它通过一个硬编码的循环（Hardcoded Loop）来运行 ReAct 逻辑，这种"黑盒"设计使得开发者很难定制复杂的执行流程，如人机交互、纠错重试：

三、数据复杂 → LangGraph

随着模型能力的一再增强，AI项目的复杂度也在不停提升，LangChain团队意识到简单的链式结构已经无法满足高级Agent开发的需求。

于是，LangGraph作为专门的编排框架应运而生。LangGraph的核心设计理念：

1. 基于图结构：使用节点（Node）、边（Edge）、状态（State）三大抽象
2. 复杂流程支持：支持循环、条件分支、并行执行
3. 状态持久化：通过检查点（checkpoint）机制实现状态保存和恢复
4. 人工介入：支持人在回路（Human-in-the-Loop）机制

只不过这依旧是过渡阶段，因为在25年模型能力真的就到位了，于是LangChain1.0才正式发布：

1.0正式版

2025年10月，LangChain 1.0和LangGraph 1.0发布，这标志着这两个框架的首个稳定版本诞生，只要1.x开头，这意味着：各位可以放心大胆用了。

后续版本的，版本稳定性会更受重视：相比早期的快速迭代，1.x 阶段更强调兼容性与迁移路径，维护成本通常更可控。并且一母同胞的边界也很清晰：

1. LangChain：更偏向应用层的组件与集成，强调易用与快速拼装
2. LangGraph：更偏向流程编排与状态管理，强调可控、可恢复、可扩展

这里也简单聊聊1.0与旧版本的区别是什么？

1.0 的重要特点

首先是整体架构方面的变化，在 1.x 生态里，一个更明显的趋势是：LangChain 更聚焦在应用层能力与集成，而 LangGraph 更适合作为流程编排与状态管理的底座能力被引入到复杂 Agent 场景中。

实际项目中二者的组合方式会因版本、语言包与团队选型而异，但整体方向是把流程控制做得更显式、更可维护。

这一架构转变，使 LangChain 从流程执行框架演进为面向开发者的应用层 SDK，带来以下显著优势：

1. 运行时能力下沉，职责边界更清晰
2. Agent 执行模型统一，减少隐式行为
3. 更强的可扩展性与可观测性
4. 为复杂 Agent 场景提供工程级稳定性保障

然后就是应用层的使用问题了，LangChain 1.0要具备不错的应用性和生态整合能力，这里他：

1. 提供高层 API 抽象，如 create_agent 等 Agent 构建接口；
2. 还内置丰富的可复用组件，包括

• agent构建
• 预构建 Chains
• Retrievers（检索器）
• Tools / Tool Calling 抽象
• Middleware（中间件/Callbacks）机制

然后是编排层，他是面向系统的 统一 Agent 运行与编排引擎，是 LangChain 1.0 的核心基础设施，他属于总控制器：

1. 作为所有复杂 Agent 与 Chain 的 统一运行时
2. 基于 状态图（State Graph） 的显式流程编排模型
3. 提供关键底层能力：

• 状态管理与持久化（Checkpointing）
• 流式响应（Streaming）
• 人工介入（Human-in-the-loop）
• 错误恢复、重试与流程回溯

最后是一张表格整理：

梳理完枯燥的历史与关键的升级后，就要回答大家实际关注的问题了：

为什么选择LangChain

很多程序员在接触 LangChain 时都会产生一个疑问：我直接用 Python 调用 OpenAI 的 API 就能实现对话，为什么要引入LangChain 这样一个复杂的框架呢？

这个时候直接上对比可能是最好的解释，比如什么场景适合手撸代码，什么场景适合使用框架？

简单来说：当你只需要实现相对简单的功能时，例如调用大模型进行文本翻译、或者构建一个基础的聊天机器人时，直接上代码就好，这个时候好处明显：

1. 完全自主可控：每一段代码的执行逻辑都清晰可见，没有额外的抽象层，整体过程透明直接。
2. 环境轻量：无需引入庞大的第三方框架，依赖简单，更适合小型或一次性任务。
3. 易于调试：可以直接查看和验证 API 返回的原始数据，问题定位直观高效。

相应的问题也很多，特别对于本身架构能力不足的团队：

1. 重复开发工作多：当功能逐渐复杂（例如需要联网搜索、保存对话历史、处理多种文件格式等），往往需要自行编写大量“粘合”代码，存在重复造轮子的情况。
2. 维护成本较高：如果需要更换模型提供商，或应对模型 API 的升级变更，通常涉及较大范围的代码调整。
3. 缺乏统一模式：在团队协作中，不同开发者对 LLM 的封装方式可能不一致，代码复用性和可维护性较差。

如果使用LangChain框架，在AI项目变得复杂后，上述问题几乎都可以规避，也就是这东西可以提升一个草台班子团队的下限，比如面对以下场景，就可以直接使用：

1. 基于检索增强生成（RAG）的问答系统（要求较高）
2. 能够自主调用外部工具（如计算器、搜索引擎）的智能助手
3. 需要在多个大模型之间灵活切换和对比效果的应用

这里的优势是：

1. 统一的模型接口：无论对接 OpenAI、Anthropic 还是 Hugging Face 等模型，LangChain 都提供一致的调用方式，通常只需调整少量配置即可完成切换。
2. 丰富的预制组件：内置文档加载、文本分割、向量存储、对话记忆等常见模块，可直接组合使用，显著提升开发效率。
3. 完善的工具集成：能够方便地接入搜索引擎、知识库以及各类业务 API，快速扩展应用能力。
4. 成熟的高阶模式支持：针对智能体（Agent）、复杂链式调用等高级场景，LangChain 提供了清晰且可复用的实现模式，减少了自行设计流程的成本。

这里依旧给一张表给大家做参考：

总结一下就是：如果你只是想体验一下 LLM，或者做一个极简的功能，直接调用 API 是最高效的；

如果你要构建一个生产级的 AI 应用，特别是涉及RAG或Agent架构，LangChain 能为你节省大量的工程化时间，让你专注于业务逻辑而非基础设施。

接下来我们来介绍下LangChain的几个核心概念（能力）：

一、create_agent

create_agent是LangChain 1.0中核心的API，它统一了智能体的创建方式。

LLM Agent通过循环运行工具来实现目标，其运行会持续直到满足停止条件，例如模型输出最终结果或达到迭代次数上限：

下面的代码展示了如何定义一个简单的工具函数，并将其绑定到 Agent 上进行调用（这里使用了 DeepSeek 模型）：

import os
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
# 1. 定义工具
@tool
defget_weather(city: str) -> str:
"""获取指定城市的天气信息"""
returnf"{city}今天天气晴朗，温度23°C"
# 2. 配置模型 (使用 DeepSeek)
llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    temperature=0
)
# 3. 创建智能体
agent = create_agent(
    model=llm,
    tools=[get_weather],
    system_prompt="你是一个专业的天气助手"
)
# 4. 执行查询
result = agent.invoke({
"messages": [{"role": "user", "content": "北京今天天气怎么样？"}]
})
# 输出结果
print(result["messages"][-1].content)
# 输出：北京今天天气晴朗，温度23°C

然后再分享个真实场景，客户服务智能体。在此场景中，我们定义了两个工具：查询订单和处理退货。Agent 会根据用户的自然语言指令，自动选择调用哪个工具：

from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI
import os
@tool
defcheck_order_status(order_id: str) -> str:
"""查询订单状态"""
# 实际应用中这里会调用数据库或API
returnf"订单{order_id}已发货，预计明天送达"
@tool
defprocess_return(order_id: str, reason: str) -> str:
"""处理退货申请"""
returnf"订单{order_id}的退货申请已受理，原因是：{reason}"
# 配置 DeepSeek 模型
llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    temperature=0
)
service_agent = create_agent(
    model=llm,
    tools=[check_order_status, process_return],
    system_prompt="""你是一个专业的客户服务助手。
    你可以帮助客户查询订单状态和处理退货申请。
    始终保持礼貌和专业的态度。"""
)
# 客户咨询
response = service_agent.invoke({
"messages": [{
"role": "user",
"content": "我的订单ORD12345发了吗？"
    }]
})
print(response["messages"][-1].content)

二、Callbacks 与 Middleware

在 LangChain 1.0 中，我们主要通过 Callbacks (回调) 和 Middleware (中间件) 来实现对 Agent 执行流程的控制与观测。

Callbacks 侧重于被动观测和简单的钩子，而 Middleware 则提供了更强大的生命周期管理和拦截能力。

Callbacks 允许开发者在智能体执行的关键节点（如 LLM 开始、结束、工具调用时）插入自定义逻辑，用于日志记录、监控或修改行为。

自定义 LoggingHandler 示例

以下示例展示了如何实现一个简单的日志回调，并注入到使用 DeepSeek 的 LLM 中：

from typing import Dict, Any, List
import os
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from langchain_openai import ChatOpenAI

classLoggingHandler(BaseCallbackHandler):
"""
    一个简单的中间件/回调，用于记录 LLM 的交互过程
    """
defon_llm_start(
        self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
    ) -> None:
"""LLM 开始处理请求时的回调"""
        print("\n[Middleware] LLM 开始处理请求...")

defon_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
"""LLM 处理完成时的回调"""
        print(f"\n[Middleware] LLM 处理完成。消耗 Token: {response.llm_output.get('token_usage', 'N/A')}")

# 使用 Callbacks 配置 DeepSeek
llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    callbacks=[LoggingHandler()]  # 注入中间件
)
# 当 Agent 使用这个 LLM 时，所有操作都会被 LoggingHandler 捕获

Middleware 提供了更强大的生命周期钩子，允许我们在模型调用前后、工具执行前后进行深度干预：

中间件生命周期钩子

1. PIIMiddleware：敏感信息保护

PII（个人身份信息）中间件可以在数据发送给大模型之前，自动识别并处理敏感信息。以下配置展示了如何自动屏蔽邮箱地址，并完全阻断包含手机号的请求：

from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware
agent = create_agent(
    model="gpt-4o-mini",
    tools=[email_tool],
    middleware=[
# 自动屏蔽邮箱地址
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
# 完全阻止电话号码传递
        PIIMiddleware(
"phone_number",
            detector=r"(?:\+?\d{1,3}[\s.-]?)?\(?\d{2,4}\)?[\s.-]?\d{3,4}[\s.-]?\d{4}",
            strategy="block"
        ),
# 屏蔽身份证号
        PIIMiddleware(
"id_card",
            detector=r"\d{17}[\dXx]",
            strategy="redact"
        )
    ]
)
# 即使用户输入包含敏感信息，也会被自动处理
result = agent.invoke({
"messages": [{"role": "user", "content": "我的邮箱是[email protected]，电话13812345678"}]
})
# 实际传递给模型的内容：我的邮箱是***，电话***

2. SummarizationMiddleware：对话历史管理

当对话轮数过多导致上下文超出模型限制时，摘要中间件会自动压缩历史记录。以下代码设置了 3000 token 的阈值，一旦超过该值就会触发自动摘要：

from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model="gpt-4o-mini",
    tools=[search_tool],
    middleware=[
        SummarizationMiddleware(
            model="gpt-4o-mini",
            max_tokens_before_summary=3000,  # 超过3000 tokens自动摘要
            summary_prompt="将以下对话内容摘要为关键要点"# 自定义摘要提示词
        )
    ]
)

# 长对话自动管理
for i in range(20):
    result = agent.invoke({
"messages": [{"role": "user", "content": f"问题{i+1}"}]
    })
# 当对话历史过长时，会自动进行摘要压缩

3. HumanInTheLoopMiddleware：人工审批机制

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

agent = create_agent(
    model="gpt-4o",
    tools=[delete_file, send_email, transfer_money],
    checkpointer=InMemorySaver(),  # 需要记忆管理支持
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
"delete_file": {
"allowed_decisions": ["approve", "edit", "reject"],
"require_reason": True# 拒绝时必须提供原因
                },
"send_email": {
"allowed_decisions": ["approve", "reject"]
                },
"transfer_money": {
"allowed_decisions": ["approve", "reject"],
"require_reason": True
                }
            }
        )
    ]
)

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

result = agent.invoke(
    {"messages": [{"role": "user", "content": "删除重要文件.doc"}]},
    config
)

if"__interrupt__"in result:
# 程序暂停，等待人工决策
    decision = input("批准删除文件？(approve/reject): ")

if decision == "approve":
# 继续执行
        result = agent.invoke(
            Command(resume={"decisions": [{"type": "approve"}]}),
            config
        )
else:
# 拒绝执行
        result = agent.invoke(
            Command(resume={
"decisions": [{
"type": "reject",
"message": "用户取消操作"
                }]
            }),
            config
        )

最后给个自定义中间件的案例：基于用户级别的动态路由

我们通过自定义中间件获取用户上下文（如会员等级），并据此动态调整使用的模型版本。以下实现展示了如何拦截请求并修改模型参数，从而为高级用户提供更强大的模型：

from dataclasses import dataclass
from typing import Callable
from langchain.chat_models import init_chat_model
from langchain.agents.middleware import AgentMiddleware
from langchain.agents.middleware.types import ModelRequest, ModelResponse

@dataclass
classContext:
    user_id: str
    user_tier: str = "free"# free, pro, enterprise
    request_count: int = 0

classTierBasedRoutingMiddleware(AgentMiddleware):
"""根据用户等级路由到不同的模型"""

def__init__(self):
        super().__init__()
# 定义不同等级使用的模型
        self.tier_models = {
"free": "gpt-4o-mini",
"pro": "gpt-4o",
"enterprise": "gpt-4o-turbo"
        }

defwrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse]
    ) -> ModelResponse:
# 获取用户等级
        user_tier = request.runtime.context.user_tier

# 动态选择模型
        model_name = self.tier_models.get(user_tier, "gpt-4o-mini")
        request.model = init_chat_model(model=model_name)

# 记录请求次数
        request.runtime.context.request_count += 1

        print(f"[中间件] 用户等级: {user_tier}, 使用模型: {model_name}")

return handler(request)

# 应用中间件
agent = create_agent(
    model="gpt-4o-mini",  # 默认模型，会被中间件覆盖
    tools=[search_tool],
    middleware=[TierBasedRoutingMiddleware()],
    context_schema=Context
)

# 不同用户使用不同模型
result_free = agent.invoke(
    {"messages": [{"role": "user", "content": "搜索新闻"}]},
    context=Context(user_id="user_001", user_tier="free")
)

result_enterprise = agent.invoke(
    {"messages": [{"role": "user", "content": "搜索新闻"}]},
    context=Context(user_id="user_002", user_tier="enterprise")
)

三、让AI返回规范数据

在实际应用中，我们需要AI返回特定格式的数据（如 JSON），而不是自然语言文本，例如：

1. 电商价格比较：返回结构化的价格列表
2. 数据提取：从文档中提取字段并返回JSON
3. API调用：需要严格按照参数格式传递

LangChain 提供了三种主要的策略来获取结构化数据，以适应不同的模型能力和场景需求。

1. AutoStrategy（推荐）：自动选择最佳策略

自动策略是获取结构化数据的最简方式。只需定义一个 Pydantic 模型，LangChain 会自动处理 Prompt 和解析逻辑，无需关心底层细节：

from langchain.agents import create_agent
from langchain.agents.structured_output import AutoStrategy
from pydantic import BaseModel

classProductComparison(BaseModel):
"""产品比较结果"""
    product_name: str
    price: float
    rating: float
    pros: list[str]
    cons: list[str]
    verdict: str  # 购买建议

agent = create_agent(
    model="gpt-4o",
    tools=[web_search],
    response_format=AutoStrategy(ProductComparison),
    system_prompt="你是一个产品比较专家"
)

result = agent.invoke({
"messages": [{
"role": "user",
"content": "比较iPhone 15和Samsung S24的优缺点"
    }]
})

# 直接获得结构化对象
comparison: ProductComparison = result["structured_response"]
print(f"产品: {comparison.product_name}")
print(f"价格: ${comparison.price}")
print(f"评分: {comparison.rating}/5")
print(f"优点: {', '.join(comparison.pros)}")
print(f"缺点: {', '.join(comparison.cons)}")
print(f"建议: {comparison.verdict}")

2. ToolStrategy：利用工具调用能力

工具策略通过模拟工具调用的方式来获取结构化输出，适用于支持 Function Calling 的模型。它利用了模型对工具参数格式的严格遵循能力：

from langchain.agents.structured_output import ToolStrategy

classWeatherForecast(BaseModel):
    city: str
    temperature: int
    condition: str  # sunny, cloudy, rainy
    humidity: int
    wind_speed: int

agent = create_agent(
    model="gpt-4o-mini",
    tools=[get_weather_data],
    response_format=ToolStrategy(WeatherForecast)
)

# 适用于任何支持工具调用的模型
# 但依赖模型自身的推理能力

3. ProviderStrategy：使用提供商原生功能

对于 OpenAI 等提供商的原生结构化输出 API（JSON mode），可以使用 ProviderStrategy 直接调用，这种方式通常比 Prompt 工程更稳定可靠：

from langchain.agents.structured_output import ProviderStrategy
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")

agent = create_agent(
    model=model,
    tools=[],
    response_format=ProviderStrategy(WeatherForecast)
)

# 直接使用OpenAI的结构化输出API
# 更稳定可靠，但仅限于支持的提供商

四、记忆管理

如果要自己去实现AI多轮对话，对架构复杂度要求是很高的，但1.0本身几个概念就直接帮一般团队将下限做了兜底：

短期记忆（对话历史）：

• 使用InMemorySaver存储在内存中
• 适合单次会话
• 程序重启后丢失

长期记忆（跨会话）：

• 使用数据库持久化（PostgreSQL、Redis等）
• 跨多次会话保持
• 需要额外的存储后端

使用 InMemorySaver 可以实现短期记忆。通过 thread_id 来区分不同的对话会话：

import os
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    temperature=0
)

# 创建带记忆的智能体
agent = create_agent(
    model=llm,
    tools=[],
    checkpointer=InMemorySaver()  # 启用内存检查点
)

# 使用thread_id区分不同会话
config = {"configurable": {"thread_id": "user_session_123"}}

# 第一轮对话
agent.invoke(
    {"messages": [{"role": "user", "content": "我叫张三，今年25岁"}]},
    config
)

# 第二轮对话（智能体记得之前的信息）
result = agent.invoke(
    {"messages": [{"role": "user", "content": "我今年多大？"}]},
    config
)
print(result["messages"][-1].content)
# 输出：你今年25岁

对于生产环境，我们需要将状态持久化到数据库。以下示例展示了如何使用 PostgreSQL 保存对话状态，即使程序重启，用户也能无缝接续之前的对话：

from langgraph.checkpoint.postgres import PostgresSaver

# 配置PostgreSQL检查点
checkpointer = PostgresSaver.from_conn_string(
"postgresql://user:password@localhost:5432/langchain"
)

agent = create_agent(
    model="gpt-4o-mini",
    tools=[search_tool],
    checkpointer=checkpointer
)

# 即使程序重启，对话历史仍然保留
config = {"configurable": {"thread_id": "user_persistent_001"}}

# 第一次运行（程序A）
result1 = agent.invoke(
    {"messages": [{"role": "user", "content": "我喜欢编程"}]},
    config
)

# 第二次运行（程序B，重启后）
result2 = agent.invoke(
    {"messages": [{"role": "user", "content": "我有什么爱好？"}]},
    config
)
# 输出：你之前提到喜欢编程

五、LangGraph

LangGraph 是一个低级别的编排框架，专门用于构建、管理和部署长期运行、有状态的 Agent。与 LangChain 提供的高级抽象不同，LangGraph 更关注底层的编排能力：

1. Durable execution（持久化执行）：Agent 可以在失败后恢复，可以长时间运行，从停止的地方继续
2. Human-in-the-loop（人工干预）：在任何点检查和修改 Agent 状态
3. Comprehensive memory（全面的记忆）：短期工作记忆 + 长期跨会话记忆
4. Debugging with LangSmith：可视化执行路径，捕获状态转换
5. Production-ready deployment：为有状态、长期运行的工作流设计的可扩展基础设施

在深入代码之前，需要理解 LangGraph 的五个核心支柱：

1. State（状态）：在所有节点之间共享的内存
2. Nodes（节点）：执行单元（Python 函数），接收状态并返回状态更新
3. Edges（边）：控制流，决定下一步去哪个节点
4. Graph（图）：编排器，将节点和边组合成可运行的工作流
5. Checkpointer（检查点）：持久化层，保存和管理执行历史

状态

状态是图的共享内存。

• 它保存了 Agent 运行过程中的所有数据（如消息历史、提取的变量、中间结果）。
• 它在节点之间传递。每个节点接收当前状态，并返回状态的更新。
• Schema 定义：通常使用 TypedDict 或 Pydantic 模型来定义状态的结构。
• 更新机制：节点返回的字典会与现有状态合并（默认行为），或者你可以定义自定义的 reducer 函数（例如追加消息列表而不是覆盖）。

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

classAgentState(TypedDict):
"""智能体状态定义"""
# 消息历史（使用add_messages自动追加而不是覆盖）
    messages: Annotated[list, add_messages]

# 自定义字段
    user_input: str
    search_results: list[dict]
    analysis_result: str
    current_step: str  # 追踪当前执行步骤

深入理解 Annotated 与 add_messages：

在上面的代码中，messages: Annotated[list, add_messages] 是一个非常关键的设计。

• 默认情况下，LangGraph 的状态更新是覆盖式的。如果节点返回 {"a": 1}，它会覆盖掉状态中原有的 a。
• 对于消息列表，我们需要的是追加而非覆盖。
• Annotated 配合 add_messages 告诉 LangGraph："当有新消息更新到这个字段时，请调用 add_messages 函数将其追加到现有列表中，而不是替换它。"

节点

节点是图的执行单元。

1. 本质上，节点就是一个 Python 函数。
2. 输入：接收当前状态（State）。
3. 输出：返回一个字典（包含要更新的状态键值对）。
4. 节点负责具体的业务逻辑，如调用 LLM、查询数据库、执行代码等。
5. 特殊节点：START（图的入口）和 END（图的出口）。

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

defsearch_node(state: AgentState, config):
"""搜索节点：执行网络搜索"""
    query = state["user_input"]

# 调用搜索工具
    results = search_web(query)

    print(f"[搜索节点] 找到 {len(results)} 条结果")

return {
"search_results": results,
"current_step": "search_completed"
    }

defanalysis_node(state: AgentState, config):
"""分析节点：分析搜索结果"""
    results = state["search_results"]

# 构造分析提示词
    prompt = f"分析以下搜索结果：\n{results}"

# 调用LLM分析
    response = llm.invoke([
        {"role": "system", "content": "你是一个信息分析专家"},
        {"role": "user", "content": prompt}
    ])

    print(f"[分析节点] 分析完成")

return {
"analysis_result": response.content,
"current_step": "analysis_completed"
    }

defreport_node(state: AgentState, config):
"""报告节点：生成最终报告"""
    analysis = state["analysis_result"]

    report = f"""
    分析报告
    =========
{analysis}
    """

    print(f"[报告节点] 报告生成完毕")

return {
"messages": [{"role": "assistant", "content": report}],
"current_step": "report_generated"
    }

边

边定义了控制流，即“下一步去哪里”：

• 普通边 (Normal Edge):START -> Node A -> Node B。这种边是确定的，A 执行完总是去 B。
• 条件边 (Conditional Edge): 根据当前状态或函数输出来动态决定下一个节点。例如，如果 LLM 决定调用工具，则跳转到 ToolNode，否则跳转到 END。
• 在 LangGraph 新版中，推荐在节点函数内部通过返回 Command(goto="next_node") 来直接控制流向，这比外部定义的条件边更直观。

from langgraph.graph import StateGraph, END

defshould_search(state: AgentState) -> str:
"""条件边：决定是否需要搜索"""
    user_input = state["user_input"]

# 如果用户问的是事实性问题，需要搜索
    question_words = ["什么", "哪里", "谁", "何时", "如何"]
if any(word in user_input for word in question_words):
return"search"

# 否则直接分析
return"analyze"

defhas_results(state: AgentState) -> str:
"""条件边：检查搜索是否有结果"""
if len(state["search_results"]) > 0:
return"analyze"
else:
return"no_results"

图

图是工作流的编排器。

• 它定义了应用程序的结构和执行逻辑。
• StateGraph 是最常用的类，它利用定义好的 State Schema 来管理数据流转。
• 编译后的图（CompiledGraph）是一个可运行的对象（Runnable），支持 invoke, stream, batch 等标准 LangChain 方法。

from langgraph.graph import StateGraph, END

# 创建状态图
builder = StateGraph(AgentState)

# 添加节点
builder.add_node("search", search_node)
builder.add_node("analyze", analysis_node)
builder.add_node("report", report_node)
builder.add_node("no_results", lambda state: {
"messages": [{"role": "assistant", "content": "未找到相关信息"}]
})

# 设置入口点
builder.set_entry_point("search")

# 添加边
# 从search节点出发，根据是否有结果决定下一步
builder.add_conditional_edges(
"search",
    has_results,
    {
"analyze": "analyze",
"no_results": "no_results"
    }
)

# analyze完成后总是到report
builder.add_edge("analyze", "report")

# report和no_results都结束
builder.add_edge("report", END)
builder.add_edge("no_results", END)

# 编译图
graph = builder.compile()

# 执行
result = graph.invoke({
"user_input": "什么是LangChain？",
"messages": [],
"search_results": [],
"analysis_result": "",
"current_step": ""
})

print(result["messages"][-1]["content"])

检查点

检查点机制是 LangGraph 的核心特性之一，它赋予了 Agent 记忆和持久化能力。

• 自动保存：它会在图执行的每个步骤（Super-step）后自动保存图的状态快照。
• 线程 (Thread): 通过 thread_id 来隔离不同的对话或执行流。
• 多级隔离：除了 thread_id，Checkpointer 还支持更细粒度的隔离（如用户级别）。在配置 configurable 时，你可以传递自定义的键值对（例如 user_id），Checkpointer 能够利用这些信息来管理和检索状态，从而实现多用户、多会话的状态管理。
• 能力解锁：
• 记忆 (Memory): 允许 Agent 跨多轮交互记住上下文。
• 人工干预 (Human-in-the-loop): 允许在特定节点暂停，等待人工批准或修改状态后再继续。
• 时间旅行 (Time Travel): 允许你查看历史执行步骤，甚至回滚到之前的某个状态，修改数据后重新执行（Fork）。
• 故障恢复：如果任务中断，可以从上次保存的状态恢复执行。

六、高级特性

复杂条件分支

通过编写路由函数，我们可以实现复杂的业务逻辑分流。例如，根据用户问题的关键词将请求分发给不同的处理节点（如客服、技术支持、销售）：

from typing import Literal

defroute_question(state: AgentState) -> Literal["web_search", "database", "direct_answer"]:
"""根据问题类型路由到不同节点"""
    question = state["user_input"].lower()

# 技术问题 -> 网络搜索
if any(word in question for word in ["如何", "怎么", "教程", "技术"]):
return"web_search"

# 数据查询问题 -> 数据库
if any(word in question for word in ["查询", "数据", "记录", "统计"]):
return"database"

# 简单问题 -> 直接回答
return"direct_answer"

builder.add_conditional_edges(
"router",
    route_question,
    {
"web_search": "web_search_node",
"database": "database_query_node",
"direct_answer": "direct_answer_node"
    }
)

循环执行

LangGraph 原生支持循环逻辑，非常适合需要反复优化或重试的场景。以下代码展示了一个最大迭代次数为 10 的循环流程，防止死循环：

classIterationState(TypedDict):
    messages: Annotated[list, add_messages]
    iteration: int
    max_iterations: int
    result: str
    converged: bool

defprocess_step(state: IterationState):
"""处理步骤"""
    iteration = state["iteration"]
    print(f"[迭代] 第 {iteration} 次处理")

# 模拟处理逻辑
    result = f"处理结果 {iteration}"

# 检查是否收敛（假设5次后收敛）
    converged = iteration >= 5

return {
"result": result,
"converged": converged,
"iteration": iteration + 1
    }

defshould_continue(state: IterationState) -> Literal["continue", "end"]:
"""决定是否继续迭代"""
if state["converged"]:
return"end"
if state["iteration"] >= state["max_iterations"]:
return"end"
return"continue"

builder = StateGraph(IterationState)
builder.add_node("process", process_step)
builder.set_entry_point("process")

# 添加循环边
builder.add_conditional_edges(
"process",
    should_continue,
    {
"continue": "process",  # 循环回process节点
"end": END
    }
)

graph = builder.compile()

# 执行（会循环5次）
result = graph.invoke({
"messages": [],
"iteration": 1,
"max_iterations": 10,
"result": "",
"converged": False
})

持久化与检查点

SQLite 是轻量级的持久化方案，适合本地开发和测试。无需额外部署数据库，只需一行代码即可启用状态保存功能：

from langgraph.checkpoint.sqlite import SqliteSaver

# 创建SQLite检查点
checkpointer = SqliteSaver.from_conn_string("agent_state.db")

builder = StateGraph(AgentState)
# ... 添加节点和边 ...

graph = builder.compile(
    checkpointer=checkpointer  # 启用检查点
)

# 使用thread_id追踪会话
config = {"configurable": {"thread_id": "user_session_001"}}

# 第一次执行
result1 = graph.invoke({"user_input": "搜索AI新闻"}, config)

# 第二次执行（从上次停止的地方继续）
result2 = graph.invoke({"user_input": "继续搜索"}, config)

在生产环境中，推荐使用 PostgreSQL 等成熟数据库。需要先建立数据库连接，再将其传递给图的编译方法，以确保高可用和数据安全：

from langgraph.checkpoint.postgres import PostgresSaver
import psycopg

# 配置PostgreSQL连接
connection = psycopg.connect(
    host="localhost",
    port=5432,
    database="langchain",
    user="postgres",
    password="password"
)

# 创建检查点
checkpointer = PostgresSaver(connection)

# 初始化表（首次运行）
checkpointer.setup()

graph = builder.compile(checkpointer=checkpointer)

LangGraph 的一大特色是“时间旅行”。我们可以查看历史状态，甚至回滚到之前的某个步骤重新执行，这对于调试和人工纠错非常有用：

# 获取会话的所有历史状态
history = graph.get_state_history(config)

# 查看历史
for state in history:
    print(f"步骤: {state.step}, 时间: {state.timestamp}")

# 回溯到特定状态
past_state = history[5]  # 回到第5步
restored_result = graph.invoke(None, config)

# 从某个历史状态重新执行
graph.update_state(config, past_state.values)
new_result = graph.invoke({"user_input": "新指令"}, config)

人工介入

在执行过程中暂停并等待人工反馈是 Agent 的常见需求。以下代码展示了如何在 human_review 节点暂停，并在收到人工指令后恢复执行：

from langgraph.types import interrupt, Command

defhuman_review_node(state: AgentState):
"""需要人工审核的节点"""
    analysis = state["analysis_result"]

# 暂停执行，等待人工输入
    feedback = interrupt({
"type": "human_review",
"content": analysis,
"question": "请审核以上分析结果"
    })

# 处理人工反馈
if feedback["approved"]:
return {"messages": [{"role": "assistant", "content": analysis}]}
else:
# 根据反馈修改
        revised = revise_analysis(analysis, feedback["comments"])
return {"messages": [{"role": "assistant", "content": revised}]}

builder.add_node("human_review", human_review_node)

执行流程：

# 第一次执行，会在human_review节点暂停
result = graph.invoke({"user_input": "分析市场趋势"}, config)

if"__interrupt__"in result:
# 获取需要审核的内容
    review_data = result["__interrupt__"]

    print("需要审核的内容：")
    print(review_data["content"])

# 人工决策
    approved = input("是否批准？(y/n): ") == "y"
    comments = ""if approved else input("请提供修改意见: ")

# 继续执行
    result = graph.invoke(
        Command(resume={
"approved": approved,
"comments": comments
        }),
        config
    )

多Agent协同

多 Agent 系统通常包含多个专门的角色。我们需要定义一个包含 current_agent 字段的共享状态，并编写路由逻辑在不同 Agent 之间流转：

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END

from langgraph.graph import add_messages, StateGraph


classMultiAgentState(TypedDict):
    messages: Annotated[list, add_messages]
    current_agent: str
    research_data: dict
    code: str
    test_results: str

defresearcher_agent(state: MultiAgentState):
"""研究Agent：负责收集信息"""
    print("[研究Agent] 开始研究...")

    research_data = {
"topic": state["messages"][-1].content,
"sources": ["source1", "source2", "source3"],
"summary": "研究摘要..."
    }

return {
"research_data": research_data,
"current_agent": "researcher"
    }

defcoder_agent(state: MultiAgentState):
"""编程Agent：负责编写代码"""
    print("[编程Agent] 开始编码...")

    research = state["research_data"]
    code = f"# 基于 {research['topic']} 生成的代码\nprint('Hello World')"

return {
"code": code,
"current_agent": "coder"
    }

deftester_agent(state: MultiAgentState):
"""测试Agent：负责测试代码"""
    print("[测试Agent] 开始测试...")

    code = state["code"]
    test_results = "测试通过！所有测试用例均通过。"

return {
"test_results": test_results,
"current_agent": "tester"
    }

defroute_to_next_agent(state: MultiAgentState) -> str:
"""路由到下一个Agent"""
    current = state["current_agent"]

if current == "researcher":
return"coder"
elif current == "coder":
return"tester"
else:
return"end"

builder = StateGraph(MultiAgentState)
builder.add_node("researcher", researcher_agent)
builder.add_node("coder", coder_agent)
builder.add_node("tester", tester_agent)

builder.set_entry_point("researcher")

# 添加路由边
builder.add_conditional_edges(
"researcher",
    route_to_next_agent,
    {"coder": "coder"}
)

builder.add_conditional_edges(
"coder",
    route_to_next_agent,
    {"tester": "tester"}
)

builder.add_conditional_edges(
"tester",
    route_to_next_agent,
    {"end": END}
)

graph = builder.compile()

# 执行多Agent协同
result = graph.invoke({
"messages": [{"role": "user", "content": "开发一个待办事项应用"}],
"current_agent": "researcher",
"research_data": {},
"code": "",
"test_results": ""
})

print("\n最终结果：")
print(f"研究数据: {result['research_data']}")
print(f"代码: {result['code']}")
print(f"测试结果: {result['test_results']}")

子图

随着 Agent 系统变得庞大，单张图会变得难以维护。LangGraph 允许我们将一个编译好的图（CompiledGraph）直接作为另一个图的节点使用。这使得我们可以像搭积木一样构建复杂系统。

# 定义子图
sub_builder = StateGraph(SubState)
# ... 构建子图 ...
sub_graph = sub_builder.compile()

# 在父图中使用子图
parent_builder = StateGraph(ParentState)
parent_builder.add_node("sub_task", sub_graph) # 直接作为节点

结语

今天我们简单的介绍了下LangChain/LangGraph框架，但这两天肾结石发作的厉害，导致我状态不好，回头再重新补一篇复杂案例吧，今天就到这...

点击上方卡片关注叶小钗公众号，查看下方二维码，添加我个人微信：