> For the complete documentation index, see [llms.txt](https://yeasy.gitbook.io/agentic_ai_guide/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://yeasy.gitbook.io/agentic_ai_guide/di-san-bu-fen-gong-cheng-shi-jian-yu-luo-di/08_frameworks/8.2_langchain.md).

# 8.2 从链到图：流程编排进化

本节用 LangChain / LangGraph 的当前官方抽象，解释智能体框架为什么会从“链式调用”走向“图编排 / 状态机”：前者适合确定性流水线，后者更适合循环、分支、检查点、容错和 Human-in-the-Loop（HITL）等生产级控制流。

## 8.2.0 过去：LCEL 适合确定性流水线

早期大家常把多个 LLM 调用按顺序串起来。这个抽象今天仍然有价值，只是它更适合 **确定性流程**，而不是需要反复决策的智能体回路。

```python
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template("将以下文本翻译成{language}: {text}")
llm = ChatOpenAI(model="gpt-5.5")
chain = prompt | llm | StrOutputParser()

result = chain.invoke({"language": "英文", "text": "你好世界"})
```

**局限**：LCEL 很适合“先做 A，再做 B”的流水线，但它不会天然提供工具循环、检查点恢复、人工审核暂停等能力。

## 8.2.1 现在的 LangChain：默认从 `create_agent` 起步

### 8.2.1.1 标准工具循环的最小写法

LangChain 官方推荐把“标准工具调用循环”先写成 `create_agent(...)`。它底层已经运行在 LangGraph 之上，所以天然具备持久化、流式输出和 HITL 扩展能力。

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

@tool
def get_order_status(order_id: str) -> str:
    """Return order status."""
    return f"{order_id}: shipped"

agent = create_agent(
    model="openai:gpt-5.5",
    tools=[get_order_status],
    system_prompt="你是客服助理，需要时先调用工具再回答。",
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "帮我查一下 ORD-001"}]}
)
print(result["messages"][-1].content)
```

这类写法适合：

* 你需要“模型决定是否调用工具”的标准 agent loop。
* 你希望先用高层抽象跑通业务，再按需下沉到底层图编排。

### 8.2.1.2 为高风险工具加上 HITL

LangChain 当前官方的 HITL 语义不是“手写一个阻塞式 `wait_for_human()` 节点”，而是：**当模型提出某个工具调用后，由 middleware 触发 interrupt，保存状态，然后用同一个 `thread_id` 恢复执行。**

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

agent = create_agent(
    model="openai:gpt-5.5",
    tools=[lookup_order, refund_order],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "lookup_order": False,
                "refund_order": {
                    "allowed_decisions": ["approve", "edit", "reject"]
                },
            }
        )
    ],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "customer-123"}}

paused = agent.invoke(
    {"messages": [{"role": "user", "content": "给 ORD-001 办理退款"}]},
    config=config,
    version="v2",
)

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

这里的关键点是：

* 中断发生在 **工具真正执行前**，而不是执行后补救。
* 恢复靠 `Command(resume=...)`，并且要复用同一个 `thread_id`。
* 决策语义是 `approve` / `edit` / `reject`，而不是随便往消息历史里塞一条“人工审核通过”。

## 8.2.2 需要显式控制流时，下沉到 LangGraph

当你需要自己定义循环、路由、子图、检查点和自定义中断逻辑时，应该直接使用 LangGraph 的 `StateGraph`。

```python
from langchain.chat_models import init_chat_model
from langchain.messages import ToolMessage
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.types import Command, interrupt

model = init_chat_model("openai:gpt-5.5", temperature=0)

@tool
def lookup_order(order_id: str) -> str:
    """Look up the order."""
    return f"{order_id}: delivered"

@tool
def refund_order(order_id: str, reason: str) -> str:
    """Submit refund request."""
    return f"{order_id}: refund requested for {reason}"

tools = [lookup_order, refund_order]
tools_by_name = {tool.name: tool for tool in tools}
model_with_tools = model.bind_tools(tools)

class CustomerState(MessagesState):
    customer_id: str

def llm_node(state: CustomerState):
    return {"messages": [model_with_tools.invoke(state["messages"])]}

def tool_or_review_node(state: CustomerState):
    results = []
    for call in state["messages"][-1].tool_calls:
        if call["name"] == "refund_order":
            decision = interrupt(
                {
                    "kind": "refund_review",
                    "tool": call["name"],
                    "args": call["args"],
                    "allowed_decisions": ["approve", "edit", "reject"],
                }
            )
            if decision["type"] == "reject":
                results.append(
                    ToolMessage(
                        content=f"人工拒绝退款：{decision['message']}",
                        tool_call_id=call["id"],
                    )
                )
                continue
            if decision["type"] == "edit":
                call = {**call, "args": decision["edited_action"]["args"]}

        tool = tools_by_name[call["name"]]
        output = tool.invoke(call["args"])
        results.append(ToolMessage(content=output, tool_call_id=call["id"]))
    return {"messages": results}

def should_continue(state: CustomerState):
    return "tool_or_review" if state["messages"][-1].tool_calls else END

builder = StateGraph(CustomerState)
builder.add_node("llm", llm_node)
builder.add_node("tool_or_review", tool_or_review_node)
builder.add_edge(START, "llm")
builder.add_conditional_edges("llm", should_continue, ["tool_or_review", END])
builder.add_edge("tool_or_review", "llm")

graph = builder.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "customer-123"}}

paused = graph.invoke(
    {
        "messages": [{"role": "user", "content": "先查询 ORD-001，必要时退款"}],
        "customer_id": "customer-123",
    },
    config=config,
)

resumed = graph.invoke(
    Command(
        resume={
            "type": "edit",
            "edited_action": {
                "name": "refund_order",
                "args": {"order_id": "ORD-001", "reason": "duplicate charge"},
            },
        }
    ),
    config=config,
)
```

这里更接近 LangGraph 官方语义：

* `interrupt()` 暂停图执行，并把一个 JSON-serializable payload 暴露给调用方。
* `Command(resume=...)` 的值会直接成为该次 `interrupt()` 的返回值。
* 图节点恢复时会从节点开头重新执行，因此中断前的副作用必须保持幂等。

## 8.2.3 高级特性

### 8.2.3.1 子图

LangGraph 支持把复杂步骤封成子图，再作为节点挂到主图里：

```python
from langgraph.graph import START, END, StateGraph

research_builder = StateGraph(ResearchState)
research_builder.add_node("search", search_node)
research_builder.add_node("summarize", summarize_node)
research_builder.add_edge(START, "search")
research_builder.add_edge("search", "summarize")
research_builder.add_edge("summarize", END)

research_graph = research_builder.compile()
main_builder.add_node("research", research_graph)
```

### 8.2.3.2 流式输出

当前官方更推荐直接使用 `stream(..., stream_mode=["updates", "messages"], version="v2")`，把 token 流和状态更新放到同一个事件流里处理：

```python
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "帮我处理退款"}]},
    config=config,
    stream_mode=["updates", "messages"],
    version="v2",
):
    if chunk["type"] == "messages":
        token, metadata = chunk["data"]
        if token.content:
            print(token.content, end="", flush=True)
    elif chunk["type"] == "updates" and "__interrupt__" in chunk["data"]:
        print("\n等待人工审核...")
```

### 8.2.3.3 节点级容错

生产环境里的图节点会遇到 LLM 限流、工具 5xx、网络抖动。LangGraph 把容错做成**节点级声明式配置**：重试、超时、以及重试耗尽后的兜底处理都直接挂在 `add_node` 上，不必在每个节点里手写退避循环和 try/except（参见 LangGraph 的[官方容错文档](https://docs.langchain.com/oss/python/langgraph/fault-tolerance)）。

```python
from langgraph.types import RetryPolicy, TimeoutPolicy
from langgraph.errors import NodeError

def on_llm_failed(state: CustomerState, error: NodeError):
    # 仅在重试全部耗尽后触发;error.node / error.error 提供失败上下文
    return {"messages": [{"role": "assistant", "content": "系统繁忙，请稍后再试。"}]}

builder.add_node(
    "llm",
    llm_node,
    retry_policy=RetryPolicy(max_attempts=4, backoff_factor=2.0),  # 指数退避重试瞬时故障
    timeout=TimeoutPolicy(run_timeout=30, idle_timeout=5),         # 墙钟超时 + 无进展超时
    error_handler=on_llm_failed,                                   # 兜底:降级回复 / 转人工 / 补偿
)
```

要点：

* `RetryPolicy` 默认只重试瞬时类异常（如 5xx、连接重置），不会重试 `ValueError` 这类代码缺陷。
* `TimeoutPolicy` 区分 `run_timeout`（单次尝试的硬墙钟上限）与 `idle_timeout`（一段时间没有可观测进展才触发），因此正常的流式节点不会被误杀。
* `error_handler` 只在重试耗尽后运行；它可以返回 `Command(goto=...)` 路由到补偿节点，从而实现 Saga 式补偿。用 `set_node_defaults(...)` 可一次性给所有节点设默认值，单个节点仍可覆盖。

## 8.2.4 小结

图编排框架的核心优势：

| 特性       | 说明                     |
| -------- | ---------------------- |
| **图结构**  | 支持循环、分支、并行             |
| **状态管理** | 类型安全的状态传递              |
| **检查点**  | 支持暂停、恢复、回放             |
| **容错**   | 节点级重试、超时与失败兜底          |
| **人机协作** | 原生支持 Human-in-the-loop |
| **可观测性** | 可与日志/指标/追踪系统集成         |

这类图编排框架适合构建生产级智能体，下一节将讨论数据/RAG 驱动的框架形态。

***

**下一节**: [8.3 数据驱动的 RAG 框架](/agentic_ai_guide/di-san-bu-fen-gong-cheng-shi-jian-yu-luo-di/08_frameworks/8.3_llamaindex.md)