LangGraph第一阶段：核心哲学与基础 (Foundations)

随着大语言模型（LLM）应用从简单的“提示词工程”转向复杂的“智能体（Agent）工作流），开发者们发现传统的线性链式结构（Chain）已经难以应对复杂的生产需求。本文将深入探讨 LangGraph 的核心哲学——状态机思维，并剖析其在生产环境中的实际应用。

一、 为什么选择 LangGraph？从 DAG 到状态机的跃迁

1. 核心痛点：DAG 无法处理“反馈环”

早期的 LLM 框架（如 LangChain 早期版本）主要基于 DAG（有向无环图） 构建。数据像流水线一样从 A 流向 B，逻辑是单向且不可逆的。

生产实际中的尴尬：
在开发一个“自动化 SQL 生成助手”时，LLM 可能会生成错误的 SQL。在 DAG 模式下，你只能：

1. 生成 SQL。
2. 运行并报错。
3. 流程结束（或者你需要写极其复杂的递归函数来处理重试）。

LangGraph 的原理：引入循环（Cycles）
LangGraph 将“流程”抽象为“状态机”。它允许节点之间存在回流，从而形成反馈环。这是构建“自愈型”智能体（Self-healing Agents）的关键。

2. 生产对比：线性链 vs. 状态机

让我们通过代码直观对比处理“重试逻辑”时的差异：

传统线性链（伪代码）：
这种结构难以处理多次失败，代码嵌套深，逻辑脆弱。

# 逻辑极其臃肿且难以扩展
result = generate_sql(prompt)
if not validate(result):
    result = fix_sql(result, error)
    if not validate(result):
        # 还要继续嵌套吗？
        ...

LangGraph 状态机实现：
通过简单的“边”定义，即可实现优雅的自动化循环。

# 定义图逻辑：如果校验失败，自动跳回生成节点
workflow.add_edge("generate", "validate")
workflow.add_conditional_edges(
    "validate",
    check_status,
    {
        "fail": "generate",  # 发现错误？跳回去重新生成！
        "success": END       # 成功？结束流程。
    }
)

3. 为什么这是前沿趋势？

• 容错力（Resilience）：生产环境中的 LLM 并非 100% 可靠。状态机允许智能体在失败时“反思”并“重试”，直至达成目标。
• 确定性控制：虽然 LLM 的输出是不确定的，但状态机的跳转逻辑是确定的。你可以严格限制智能体最多重试 3 次，或者在特定错误下必须请求人工介入。
• 与 LCEL 深度融合：LangGraph 继承了 LangChain Expression Language 的声明式优势，同时补齐了其在复杂逻辑控制上的短板。

二、 核心组件：构建智能体的三根支柱

在 LangGraph 中，一切逻辑都围绕着“图”展开。理解这三根支柱，才能真正从“写 Prompt”进阶到“写系统”。

1. State (状态) —— 智能体的“全域持久化记忆”

状态是整个图的共享上下文。在 LangGraph 中定义状态，就像是在制定一份“团队协作契约”。

① TypedDict：数据契约（我是谁？我有啥？）

• 原理：它是 Python 的一种类型标注，规定了字典里必须有哪些“键”以及“值”的类型。
• 为什么用：在复杂的图中，数据会在几十个节点间流转。如果没有 TypedDict，节点 A 可能会写 msg，节点 B 却在读 messages，导致系统崩溃。它提供了 IDE 代码补全和静态检查，确保所有节点对数据的理解完全一致。
• 直观类比：就像是一个“公用文件夹”的封面目录，谁存了什么，谁该取什么，一目了然。

② Annotated：更新协议（怎么存？覆盖还是累加？）

• 原理：它是 Python 的元数据标注。在 LangGraph 中，它用来绑定 Reducer（归约器） 函数（如 operator.add）。
• 为什么用：
  • 普通变量：赋值即覆盖（a = 5, a = 6，结果是 6）。
  • 智能体状态：我们往往需要“累加”历史（比如对话记录）。
  • Annotated[list, operator.add] 告诉 LangGraph：“当有新数据进来时，请调用 add 方法把它拼接到旧数据后面，而不是把旧的删掉。”
• 直观类比：普通赋值像是在黑板上写字（擦掉旧的写新的）；Annotated 像是在记事本上写字（在上一行后面接着写）。

③ Sequence：泛型序列（我接收任何有序列表）

• 原理：它是一个抽象的类型标注，表示“一组有序的东西”。
• 为什么用：
  • 比起具体的 list，Sequence 更健壮。它能兼容 list、tuple 等多种有序容器。
  • 在处理 LangChain 的消息流时，使用 Sequence[BaseMessage] 是一种行业标准，确保你的智能体可以无缝处理来自不同模型的各种消息类型（Human, AI, System, Tool）。

综合代码示例：

from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
import operator

class GraphState(TypedDict):
    # 核心：定义消息如何流动
    # 每次节点返回消息时，都会自动 append 到这个 Sequence 中
    messages: Annotated[Sequence[BaseMessage], operator.add]
    
    # 业务：这就是普通的覆盖更新（最新的状态覆盖旧的）
    current_task_status: str 
    is_safe: bool

2. Nodes (节点) —— 独立的“职能逻辑单元”

节点是图中的处理环节。它接收当前状态，执行操作，然后输出更新后的状态。

• 原理：每个节点都是一个普通的 Python 函数。
• 生产实践：将推理、工具执行和人工审计解耦为不同的节点。

def research_node(state: GraphState):
    """专门负责检索文档的节点"""
    last_message = state["messages"][-1]
    # 模拟检索逻辑
    docs = search_vector_db(last_message.content)
    # 节点只返回需要更新的字段，其余字段由系统自动合并
    return {"user_context": {"docs": docs}}

3. Edges (边) —— 智能体的“决策路由”

边决定了状态流转的方向。

• 普通边：硬编码的固定路径。
• 条件边：智能体的“思考”体现。根据当前状态内容，决定下一步是去调用工具（Action）还是直接回答（End）。

def router(state: GraphState):
    """分析 LLM 回复以决定路径"""
    if "ERROR" in state["messages"][-1].content:
        return "retry"
    return "continue"

三、 生产实际：ReAct 模式下的自动化客服

场景背景与痛点：
假设我们要开发一个电商售后助理。用户问：“我的订单 1024 什么时候到？”。

• 挑战：LLM 无法预知实时物流。如果只靠 Prompt，它可能会产生严重的幻觉（如随意回复“明天就到”）。
• 方案：采用 ReAct (Reasoning and Acting) 架构，将 LLM 变成“思考中枢”，将 API 变成“手脚”。

ReAct 的生产工作流逻辑：

1. Reasoning (推理)：LLM 接收到消息，分析状态后返回一个工具调用意图：“我需要调用 get_logistics API，参数是 order_id=1024”。
2. Acting (行动)：LangGraph 将流向转至 Action 节点，执行真实的 HTTP 请求。
3. Observation (观察)：系统将 API 返回的真实数据（如：“由于大雪，预计延迟两天”）写回 State。
4. 循环决策：状态再次回到 LLM。它看到“观察”到的结果，进行第二次推理：“物流确实延迟了，我现在可以如实告诉用户了”。

为什么用 LangGraph 实现 ReAct？
在实际生产中，API 可能会超时或报错。在 LangGraph 的闭环中，如果 API 报错，我们可以引导 LLM 看到报错信息并尝试自愈（例如改用更简单的查询参数再试一次），或者通过条件边优雅地走向“转人工服务”。这种“思考-行动-反馈”的闭环，是构建高可靠 AI 系统的唯一路径。

四、 快速上手：构建第一个 ReAct 代理

from langgraph.graph import StateGraph, END
# ...（此处省略部分 import）

# 1. 定义状态 (使用我们学到的 TypedDict)
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

# 2. 定义节点
model = ChatOpenAI(temperature=0)

def call_model(state: AgentState):
    """思考节点：决定下一步做什么"""
    response = model.invoke(state['messages'])
    return {"messages": [response]}

# 3. 构建工作流
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.set_entry_point("agent")

# 设置条件流转：这是 ReAct 的核心，根据模型输出决定是去执行工具还是结束
workflow.add_conditional_edges(
    "agent",
    should_continue, # 这个函数检查最后一条消息是否包含 tool_calls
    {"continue": "action", "end": END}
)

# 行动完成后必须回到 agent 节点进行新一轮的 Observation 思考
workflow.add_edge("action", "agent")

五、 总结：为什么它是“构建可控智能体”的基石？

1. 容错性：节点失败可重试。
2. 人机交互：支持中途断点人工介入。
3. 可视化：逻辑链路透明，业务可审计。

通过掌握这些核心组件，你已经完成了从“写 Prompt”到“设计 AI 系统”的华丽转身。