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

本节通过一个典型生态的演进路径，解释智能体框架为何从“链式调用”走向“图编排/状态机”：前者上手快，后者更适合表达循环、分支、回退与检查点等生产级控制流。

8.2.1 从链式调用到图编排

早期：链式编排

早期框架的核心抽象常是 链（Chain）——将多个 LLM 调用串联起来：

# 早期的链式调用

from langchain import LLMChain, PromptTemplate

chain = LLMChain(
    llm=llm,
    prompt=PromptTemplate.from_template("将以下文本翻译成{language}: {text}")
)

result = chain.run(language="英文", text="你好世界")

问题：Chain 是线性的，无法表达分支、循环和复杂的控制流。

中期：引入智能体执行器

为了支持智能体的 ReAct 循环，一些框架引入了“智能体执行器”抽象：

from langchain.agents import AgentExecutor, create_react_agent

agent = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools)

result = executor.invoke({"input": "今天北京天气如何？"})

问题：执行器往往是“黑盒”，难以定制控制流、添加中间步骤或实现人工审核节点。

现在：图编排/状态机

图编排把工作流建模为 有向图，提供更强的控制力：

# 伪代码示例：用“图编排”构建带人工审核的客服智能体

# 1) 定义状态（包含消息、用户标识、是否需要人工审核等字段）
class CustomerServiceState(TypedDict):
    messages: list
    customer_id: str
    requires_human: bool

# 2) 定义工具（查询订单、申请退款等）
tools = [get_order_status, request_refund]

# 3) 定义节点（智能体节点、工具执行节点、人工审核节点）
def agent_node(state):
    response = llm.decide_next(messages=state["messages"], tools=tools)
    requires_human = any(call.name == "request_refund" for call in response.tool_calls)
    return {"messages": state["messages"] + [response], "requires_human": requires_human}

def tools_node(state):
    tool_results = execute_tool_calls(state["messages"][-1].tool_calls)
    return {"messages": state["messages"] + tool_results}

def human_review_node(state):
    # 在此暂停，等待人类批准/拒绝/修改参数
    decision = wait_for_human()
    return apply_human_decision(state, decision)

# 4) 构建图：agent -> (tools | human_review | end)
app = Graph(State=CustomerServiceState)
app.add_node("agent", agent_node)
app.add_node("tools", tools_node)
app.add_node("human_review", human_review_node)
app.add_conditional_edges("agent", router=route)
app.add_edge("tools", "agent")
app.add_edge("human_review", "agent")
app.enable_checkpoint_and_interrupt(before=["human_review"])  # 支持中断/恢复

使用示例

具体示例如下：

# 启动对话

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

# 第一轮：查询订单

response = app.invoke(
    {"messages": [HumanMessage(content="我的订单 ORD-001 到哪了？")]},
    config
)
print(response["messages"][-1].content)
# 输出：订单 ORD-001 状态：已发货，预计明天送达

# 第二轮：申请退款（触发人工审核）

response = app.invoke(
    {"messages": [HumanMessage(content="我要退货")]},
    config
)
# 图在 human_review 节点前暂停

# 人工审核后继续

approval = HumanMessage(content="[人工审核] 批准退款申请")
response = app.invoke({"messages": [approval]}, config)

8.2.4 高级特性

子图

将复杂逻辑封装为可复用的子图：

# 定义子图

def create_research_subgraph():
    subgraph = StateGraph(ResearchState)
    subgraph.add_node("search", search_node)
    subgraph.add_node("summarize", summarize_node)
    # ...

    return subgraph.compile()

# 在主图中使用

main_graph.add_node("research", create_research_subgraph())

流式输出

具体示例如下：

async for event in app.astream_events(input, config, version="v2"):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="", flush=True)

8.2.5 小结

图编排框架的核心优势：

特性	说明
图结构	支持循环、分支、并行
状态管理	类型安全的状态传递
检查点	支持暂停、恢复、回放
人机协同	原生支持 Human-in-the-loop
可观测性	可与日志/指标/追踪系统集成

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

---

下一节: 8.3 数据驱动的 RAG 框架