# 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` 起步

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.1 为高风险工具加上 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 MemorySaver
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=MemorySaver())

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.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)


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://yeasy.gitbook.io/agentic_ai_guide/di-san-bu-fen-gong-cheng-shi-jian-yu-luo-di/08_frameworks/8.2_langchain.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.