LangGraph 15：工作流模式

前言

把节点与边拼起来之后，真正复用的是工作流形状：串行精炼、扇出扇入、结构化路由、动态工人、评估闭环、工具 Agent。
本文作本系列收束，每种模式给一段可跑骨架，方便对照改造成业务图。
示例统一对接 火山方舟 Coding Plan，模型用 ark-code-latest。
下文需要 Python 3.12+，依赖用 uv 管理。

概要

本文用短示例串六种常见工作流形状，作为系列收束对照。
依次是提示链、并行化、路由、编排与工人、评估优化、工具 Agent。
每种只给可跑骨架，细节回前面各篇。

依赖

建议使用 Python 3.12 及以上。

uv init langgraph-workflow-patterns
cd langgraph-workflow-patterns
uv venv --python 3.12
uv add "langgraph>=1.0,<2.0" "langchain>=1.0,<2.0" langchain-openai python-dotenv pydantic rich

在项目根目录配置 Coding Plan 的 .env。
下文各节默认共用同一模型初始化方式，可抽到公共模块。

提示链

多步 LLM 串行：生成 → 门控 → 改进 → 润色。
条件边决定「已经够好则直接结束」，否则继续精炼。

import os
from typing import Literal, TypedDict

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langgraph.graph import END, START, StateGraph
from rich import print as rprint

load_dotenv()
model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0.7,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)

class OverAllState(TypedDict):
    topic: str
    joke: str
    improved_joke: str
    final_joke: str

def generate_joke(state: OverAllState) -> OverAllState:
    msg = model.invoke([HumanMessage(content=f"写一个关于「{state['topic']}」的短笑话")])
    return {"joke": msg.content}

def check_punchline(state: OverAllState) -> Literal["pass", "no_pass"]:
    if "?" in state["joke"] or "!" in state["joke"] or "？" in state["joke"] or "！" in state["joke"]:
        return "pass"
    return "no_pass"

def improve_joke(state: OverAllState) -> OverAllState:
    msg = model.invoke(
        [HumanMessage(content=f"给下面笑话加一点双关，使其更有趣：\n{state['joke']}")]
    )
    return {"improved_joke": msg.content}

def polish_joke(state: OverAllState) -> OverAllState:
    msg = model.invoke(
        [HumanMessage(content=f"为下面笑话加一个反转：\n{state['improved_joke']}")]
    )
    return {"final_joke": msg.content}

builder = StateGraph(state_schema=OverAllState)
builder.add_node("generate_joke", generate_joke)
builder.add_node("improve_joke", improve_joke)
builder.add_node("polish_joke", polish_joke)
builder.add_edge(START, "generate_joke")
builder.add_conditional_edges(
    "generate_joke",
    check_punchline,
    {"no_pass": "improve_joke", "pass": END},
)
builder.add_edge("improve_joke", "polish_joke")
builder.add_edge("polish_joke", END)
graph = builder.compile()
rprint(graph.invoke({"topic": "猫"}))

并行化

从 START 扇出多个互不依赖的 LLM 节点，再用「多源边」汇入聚合节点。

import os
from typing import TypedDict

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langgraph.graph import END, START, StateGraph
from rich import print as rprint

load_dotenv()
model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0.7,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)

class OverAllState(TypedDict):
    topic: str
    joke: str
    story: str
    poem: str
    combined_output: str

def call_joke(state: OverAllState) -> OverAllState:
    msg = model.invoke([HumanMessage(content=f"写一个关于「{state['topic']}」的短笑话")])
    return {"joke": msg.content}

def call_story(state: OverAllState) -> OverAllState:
    msg = model.invoke([HumanMessage(content=f"写一个关于「{state['topic']}」的短故事")])
    return {"story": msg.content}

def call_poem(state: OverAllState) -> OverAllState:
    msg = model.invoke([HumanMessage(content=f"写一首关于「{state['topic']}」的短诗")])
    return {"poem": msg.content}

def aggregator(state: OverAllState) -> OverAllState:
    text = (
        f"主题：{state['topic']}\n\n"
        f"故事：\n{state['story']}\n\n"
        f"笑话：\n{state['joke']}\n\n"
        f"诗歌：\n{state['poem']}"
    )
    return {"combined_output": text}

builder = StateGraph(state_schema=OverAllState)
builder.add_node("call_joke", call_joke)
builder.add_node("call_story", call_story)
builder.add_node("call_poem", call_poem)
builder.add_node("aggregator", aggregator)
builder.add_edge(START, "call_joke")
builder.add_edge(START, "call_story")
builder.add_edge(START, "call_poem")
builder.add_edge(["call_joke", "call_story", "call_poem"], "aggregator")
builder.add_edge("aggregator", END)
graph = builder.compile()
rprint(graph.invoke({"topic": "猫"})["combined_output"])

路由

用 with_structured_output 让模型只输出枚举步骤，再条件边分发到对应生成节点。

import os
from typing import Literal, TypedDict

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage, SystemMessage
from langgraph.graph import END, START, StateGraph
from pydantic import BaseModel, Field
from rich import print as rprint

load_dotenv()
model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)

class Route(BaseModel):
    step: Literal["poem", "story", "joke"] = Field(description="下一执行步骤")

router_llm = model.with_structured_output(Route)

class OverAllState(TypedDict):
    input: str
    decision: str
    output: str

def model_call_router(state: OverAllState) -> OverAllState:
    decision = router_llm.invoke(
        [
            SystemMessage(
                content="根据请求路由到 story、joke 或 poem。"
            ),
            HumanMessage(content=state["input"]),
        ]
    )
    return {"decision": decision.step}

def write_story(state: OverAllState) -> OverAllState:
    return {"output": model.invoke([HumanMessage(content=state["input"])]).content}

def write_joke(state: OverAllState) -> OverAllState:
    return {"output": model.invoke([HumanMessage(content=state["input"])]).content}

def write_poem(state: OverAllState) -> OverAllState:
    return {"output": model.invoke([HumanMessage(content=state["input"])]).content}

def route_decision(state: OverAllState) -> Literal["write_story", "write_joke", "write_poem"]:
    mapping = {"story": "write_story", "joke": "write_joke", "poem": "write_poem"}
    return mapping[state["decision"]]

builder = StateGraph(state_schema=OverAllState)
builder.add_node("model_call_router", model_call_router)
builder.add_node("write_story", write_story)
builder.add_node("write_joke", write_joke)
builder.add_node("write_poem", write_poem)
builder.add_edge(START, "model_call_router")
builder.add_conditional_edges(
    "model_call_router",
    route_decision,
    {
        "write_story": "write_story",
        "write_joke": "write_joke",
        "write_poem": "write_poem",
    },
)
builder.add_edge("write_story", END)
builder.add_edge("write_joke", END)
builder.add_edge("write_poem", END)
graph = builder.compile()
rprint(graph.invoke({"input": "写一个关于猫的诗"})["output"])

编排与工人

编排器先规划章节列表，再用 Send("worker", payload) 动态扇出不确定数量的工人；工人结果用 Annotated[list, add] 归并，最后合成报告。

import os
from operator import add
from typing import Annotated, List, TypedDict
from collections.abc import Sequence

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage, SystemMessage
from langgraph.graph import END, START, StateGraph
from langgraph.types import Send
from pydantic import BaseModel, Field
from rich import print as rprint

load_dotenv()
model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)

class Section(BaseModel):
    name: str = Field(description="章节名称")
    description: str = Field(description="本章概述")

class Sections(BaseModel):
    sections: List[Section]

planner = model.with_structured_output(Sections)

class OverAllState(TypedDict):
    topic: str
    sections: list[Section]
    completed_sections: Annotated[list, add]
    final_report: str

class WorkerState(TypedDict):
    section: Section
    completed_sections: Annotated[list, add]

def orchestrator(state: OverAllState) -> OverAllState:
    plan = planner.invoke(
        [
            SystemMessage(content="为报告生成章节规划。"),
            HumanMessage(content=f"主题：{state['topic']}"),
        ]
    )
    return {"sections": plan.sections}

def worker(state: WorkerState) -> WorkerState:
    section = model.invoke(
        [
            SystemMessage(content="按章节名与描述写 Markdown 正文，不要额外开场白。"),
            HumanMessage(
                content=f"名称：{state['section'].name}\n描述：{state['section'].description}"
            ),
        ]
    )
    return {"completed_sections": [section.content]}

def synthesizer(state: OverAllState) -> OverAllState:
    body = "\n\n".join(state["completed_sections"])
    return {"final_report": body}

def assign_workers(state: OverAllState) -> Sequence[Send]:
    return [Send("worker", {"section": s}) for s in state["sections"]]

builder = StateGraph(state_schema=OverAllState)
builder.add_node("orchestrator", orchestrator)
builder.add_node("worker", worker)
builder.add_node("synthesizer", synthesizer)
builder.add_edge(START, "orchestrator")
builder.add_conditional_edges("orchestrator", assign_workers, ["worker"])
builder.add_edge("worker", "synthesizer")
builder.add_edge("synthesizer", END)
graph = builder.compile()
out = graph.invoke({"topic": "大语言模型缩放定律简介"})
rprint(out["final_report"])

评估优化

生成器与评估器闭环：评估器结构化输出「通过 / 不通过 + 反馈」，不通过则带着反馈再生成。
建议设置 recursion_limit，避免死循环。

import os
from typing import Literal, TypedDict

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langgraph.graph import END, START, StateGraph
from pydantic import BaseModel, Field
from rich import print as rprint

load_dotenv()
model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0.7,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)

class Feedback(BaseModel):
    grade: Literal["好笑", "不好笑"] = Field(description="是否好笑")
    feedback: str = Field(description="改进建议")

evaluator = model.with_structured_output(Feedback)

class OverAllState(TypedDict):
    joke: str
    topic: str
    feedback: str
    funny_or_not: str

def generator(state: OverAllState) -> OverAllState:
    if state.get("feedback"):
        prompt = f"写一个关于「{state['topic']}」的笑话。参考建议：{state['feedback']}"
    else:
        prompt = f"写一个关于「{state['topic']}」的笑话"
    return {"joke": model.invoke(prompt).content}

def evaluate(state: OverAllState) -> OverAllState:
    grade = evaluator.invoke(f"评估下面笑话是否好笑：\n{state['joke']}")
    return {"funny_or_not": grade.grade, "feedback": grade.feedback}

def route_joke(state: OverAllState) -> Literal["accept", "reject_and_feedback"]:
    return "accept" if state["funny_or_not"] == "好笑" else "reject_and_feedback"

builder = StateGraph(state_schema=OverAllState)
builder.add_node("generator", generator)
builder.add_node("evaluate", evaluate)
builder.add_edge(START, "generator")
builder.add_edge("generator", "evaluate")
builder.add_conditional_edges(
    "evaluate",
    route_joke,
    {"accept": END, "reject_and_feedback": "generator"},
)
graph = builder.compile()
state = graph.invoke({"topic": "猫"}, config={"recursion_limit": 5})
rprint(state["joke"])

工具 Agent

最小 ReAct 形：模型节点 ↔ ToolNode，条件边看是否还有 tool_calls。
完整讲解见 《LangGraph 06：工具调用》，本节只给可对照的短骨架。

import os
from typing import Literal

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

load_dotenv()

@tool(parse_docstring=True)
def get_weather(city: str) -> str:
    """根据城市查询当日天气。

    Args:
        city: 城市名称
    """
    return f"{city} 今天天气不错"

tools = [get_weather]
model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
).bind_tools(tools)

def model_node(state: MessagesState) -> MessagesState:
    return {"messages": [model.invoke(state["messages"])]}

def router(state: MessagesState) -> Literal["tool_node", "__end__"]:
    return "tool_node" if state["messages"][-1].tool_calls else END

builder = StateGraph(state_schema=MessagesState)
builder.add_node("model_node", model_node)
builder.add_node("tool_node", ToolNode(tools=tools))
builder.add_edge(START, "model_node")
builder.add_conditional_edges("model_node", router, path_map=["tool_node", END])
builder.add_edge("tool_node", "model_node")
graph = builder.compile()

res = graph.invoke({"messages": [HumanMessage(content="今天北京天气怎么样？")]})
for msg in res["messages"]:
    msg.pretty_print()

需要人审工具时，把 《LangGraph 11：HITL 人机协同》 的 interrupt 嵌进工具或自定义 tool_node 即可。

总结

1. 提示链：串行精炼 + 门控提前结束。
2. 并行：扇出多 LLM，多源边汇聚。
3. 路由：结构化枚举 + 条件边。
4. 编排-工人：Send 动态并行，Annotated[..., add] 归并。
5. 评估-优化：生成 ↔ 评估闭环，限制递归。
6. 工具 Agent：bind_tools + ToolNode 构成最小可用环。