跳到正文

TECH NOTE

2397 字 12 分钟 系列 8 / 22

08 Custom workflow 自定义工作流：流程由你画，该用代码的地方就用代码

多 Agent LangChain 多Agent
收录于合集 多 Agent 精讲 / 五种模式
合集目录
ON THIS PAGE / 目录9 节

一句话讲透：前四种模式都是”用模型决定路由”；Custom workflow 是”路由由你决定，模型只干该干的活”。用 LangGraph 把流程画成图：哪一步是普通函数、哪一步是一次 LLM 调用、哪一步是一个完整 Agent，全由你定。

它到底是什么#

原文定义：

在 custom workflow 架构里，你用 LangGraph 定义自己的定制执行流。你对图结构有完全的控制权——包括顺序步骤、条件分支、循环和并行执行。

graph LR
    A([输入]) --> B{{条件分支}}
    B -->|path_a| C[确定性步骤]
    B -->|path_b| D((Agent 步骤))
    C --> G([输出])
    D --> G

这张图在说什么：① 左边进来先做一次条件判断（可以是普通代码，也可以是一次模型调用）；② 两条路性质不同：path_a 走的是确定性步骤（纯代码，比如查库、算数、格式转换），path_b 走的是Agent 步骤（带工具、要推理）；③ 两条路最后汇到同一个输出。这就是 Custom workflow 的精髓：按需混用”代码”和”模型”。

四个关键特征#

特征原文说人话
完全控制图结构Complete control over graph structure顺序、分支、循环、并行，随你连
混合确定性与 Agent 行为Mix deterministic logic with agentic behavior能用代码的地方就别用模型（省钱、稳定、可测）
支持顺序/分支/循环/并行Sequential steps, conditional branches, loops, parallel execution图能表达的东西它都能表达
可以嵌套其它模式Embed other patterns as nodes in your workflow把一整个多 Agent 系统塞进一个节点

原文对”节点里能放什么”说得很清楚：

你的工作流里每个节点都可以是一个简单函数、一次 LLM 调用，或者一个带工具的完整 Agent。你还可以在工作流里组合其它架构——例如把一整个多 Agent 系统嵌成一个节点。

🧭 补充解读（非原文）：这一条是本篇最值钱的设计原则，和工程常识完全一致—— “该确定的事，永远不要交给模型”。校验、格式转换、金额计算、权限判断、字段映射：写成节点函数，可测、可复现、几乎零成本。模型只留给真正需要推理/判断的节点。 反过来说，如果你发现整个流程用 if/match + 几个函数就能写清楚，那就别用 LangGraph（这一点在第 01 篇也提过）。

什么时候用它#

原文三条：

  • 标准模式（subagents、skills 等）都不合适；
  • 你需要把确定性逻辑和 Agent 行为混在一起；
  • 你的场景需要复杂的路由或多阶段处理。

📎 官方实战教程：Build a multi-source knowledge base with routing。 原文明确说了：Router 模式本身就是 Custom workflow 的一个例子——所以 07 篇的实战教程，同时也是本篇最好的完整案例（本笔记的精讲版见 12）。

最小实现：在任何 LangGraph 节点里直接调 Agent#

原文点出了核心洞察：

你可以在任何 LangGraph 节点里直接调用 LangChain Agent——这样就同时拿到了自定义工作流的灵活性和预置 Agent 的便利性。

from langchain.agents import create_agent
from langgraph.graph import StateGraph, START, END


agent = create_agent(model="openai:gpt-5.5", tools=[...])


def agent_node(state: State) -> dict:
    """一个 LangGraph 节点：里面调用一个 LangChain Agent。"""
    result = agent.invoke({
        "messages": [{"role": "user", "content": state["query"]}]
    })
    return {"answer": result["messages"][-1].content}


# 组一个最简单的工作流
workflow = (
    StateGraph(State)
    .add_node("agent", agent_node)
    .add_edge(START, "agent")
    .add_edge("agent", END)
    .compile()
)

这段代码在说什么：① create_agent 造出来的 Agent 就是一个普通对象，可以在任何函数里 .invoke(...)；② 节点函数的契约是”读 state → 返回要更新的字段”（这里是 {"answer": ...}）；③ 用 StateGraph 把节点连起来，compile() 之后就是一个可执行的工作流。“Agent 即节点”是这套架构的全部魔法。

实例：一个 RAG 管道（三种节点全用上）#

原文给了一个 WNBA 数据助手的例子：先从知识库检索，还能去取最新新闻。它演示了三类节点：

节点类型例子里的节点干什么
模型节点（Model node）Rewrite用结构化输出把用户问题改写成更适合检索的查询
确定性节点（Deterministic node）Retrieve做向量相似度检索——完全不涉及 LLM
Agent 节点（Agent node）Agent基于检索到的上下文推理，还能通过工具去取额外信息
graph LR
    A([用户问题]) --> B{{改写 Rewrite}}
    B --> C[(检索 Retrieve)]
    C --> D((Agent))
    D --> E([回答])

这张图在说什么：① “改写”是一次便宜的模型调用（只为提升检索命中率）；② “检索”是纯代码，没有任何模型参与；③ “推理 + 可能调工具”才交给完整 Agent。整条链路只在必要处花钱。

💡 提示（原文）：你可以用 LangGraph 的 state 在工作流各步之间传信息。这让流程里的每一部分都能读写结构化字段，从而方便地在节点之间共享数据与上下文。

完整代码（原文，已精简）#

from typing import TypedDict
from pydantic import BaseModel
from langgraph.graph import StateGraph, START, END
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore


class State(TypedDict):
    question: str
    rewritten_query: str
    documents: list[str]
    answer: str


# —— 知识库与检索器（内容略，原文是 WNBA 名单/赛果/球员数据）——
embeddings = OpenAIEmbeddings()
vector_store = InMemoryVectorStore(embeddings)
vector_store.add_texts([...])
retriever = vector_store.as_retriever(search_kwargs={"k": 5})


@tool
def get_latest_news(query: str) -> str:
    """Get the latest WNBA news and updates."""
    # 这里接你自己的新闻 API
    return "Latest: The WNBA announced expanded playoff format for 2025..."


agent = create_agent(model="openai:gpt-5.5", tools=[get_latest_news])
model = ChatOpenAI(model="gpt-5.5")


class RewrittenQuery(BaseModel):
    query: str


# ① 模型节点：改写查询（结构化输出）
def rewrite_query(state: State) -> dict:
    """把用户问题改写成更适合检索的查询。"""
    system_prompt = """Rewrite this query to retrieve relevant WNBA information.
The knowledge base contains: team rosters, game results with scores, and player statistics (PPG, RPG, APG).
Focus on specific player names, team names, or stat categories mentioned."""
    response = model.with_structured_output(RewrittenQuery).invoke([
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": state["question"]},
    ])
    return {"rewritten_query": response.query}


# ② 确定性节点：纯向量检索，无 LLM
def retrieve(state: State) -> dict:
    """基于改写后的查询检索文档。"""
    docs = retriever.invoke(state["rewritten_query"])
    return {"documents": [doc.page_content for doc in docs]}


# ③ Agent 节点：带着检索结果推理，还能调工具
def call_agent(state: State) -> dict:
    """用检索到的上下文生成回答。"""
    context = "\n\n".join(state["documents"])
    prompt = f"Context:\n{context}\n\nQuestion: {state['question']}"
    response = agent.invoke({"messages": [{"role": "user", "content": prompt}]})
    return {"answer": response["messages"][-1].content_blocks}


workflow = (
    StateGraph(State)
    .add_node("rewrite", rewrite_query)
    .add_node("retrieve", retrieve)
    .add_node("agent", call_agent)
    .add_edge(START, "rewrite")
    .add_edge("rewrite", "retrieve")
    .add_edge("retrieve", "agent")
    .add_edge("agent", END)
    .compile()
)


result = workflow.invoke({"question": "Who won the 2024 WNBA Championship?"})
print(result["answer"])

这段代码在说什么：① State 是贯穿全流程的结构化数据袋（问题 → 改写后查询 → 检索到的文档 → 回答）；② 三种节点的写法完全一样（都是 state -> dict），区别只在里面干了什么；③ .add_edge() 把这些节点串成一条直线——要加分支就换 add_conditional_edges，要加循环就把边连回去，要并行就扇出（见 07 和 12）。

ℹ️ 生产环境提醒（原文）：别用 InMemoryVectorStore，请换成持久化的向量库，例如 Valkey、Databricks Vector Search 或 MongoDB Atlas。完整清单见所有向量库。

⚠️ 原文如此：call_agent 里返回的是 response["messages"][-1].content_blocks（内容块列表），而 State 里声明的 answer: str。原文就是这么写的，类型上并不严格一致；自己落地时建议统一（取 .content 或把 State 改成块列表）。

Custom workflow 与前四种模式的关系#

        ┌───────────────────── 谁来决定"下一步去哪" ─────────────────────┐
        │                                                              │
   模型决定（Agentic）                                    你决定（Deterministic）
        │                                                              │
   ┌────┴─────┬──────────┬───────────┐                    ┌─────────────┴─────────────┐
 Subagents  Handoffs   Skills     Router              Custom workflow（LangGraph）
 （主管）    （状态机） （按需加载）（一次分诊）        顺序/分支/循环/并行 由代码连
        │         │          │           │                    │
        └─────────┴──────────┴───────────┘                    │
                       │                                     │
                       └────── 都可以被"塞进" ────────────────┘
                              Custom workflow 的一个节点里

🧭 补充解读（非原文）：把这张图记住，多 Agent 选型就不会乱——

  • 需要模型临场决策（派谁、换谁、加载什么）→ 前四种模式。
  • 需要流程可控（必须先 A 再 B、失败要重试、要循环直到达标）→ Custom workflow。
  • 两者不是竞争关系：生产系统常见形态是”LangGraph 主干 + 局部 Agent 节点”——主干保证流程与可观测性，Agent 节点提供灵活判断。

小结#

  1. 一句话：Custom workflow = 用 LangGraph 自己画图，节点可以是普通函数、一次 LLM 调用，或一个完整 Agent。
  2. 四个特征：完全控制图结构、混合确定性与 Agent 行为、支持顺序/分支/循环/并行、可以把其它模式当作节点嵌进来。
  3. 核心洞察：在任何节点里都能直接 .invoke() 一个 LangChain Agent——“Agent 即节点”。
  4. RAG 实例演示了三类节点：模型节点（改写）、确定性节点（检索）、Agent 节点（推理 + 工具），这正是”该花的花、该省的省”。
  5. State 是粘合剂：节点之间靠结构化的图状态传数据（TypedDict + 每个节点返回要更新的字段）。
  6. 它是最贴近生产形态的模式：流程与副作用由代码掌控，只有真正需要推理的位置才放模型。

一句话选型#

  • 选它：标准模式都不合适；需要把确定性逻辑与 Agent 混用；流程有分支/循环/重试/多阶段；需要把整个多 Agent 系统当成一个可观测、可恢复的图来跑。
  • 别选它：需求正好命中前四种模式（别再手搓一遍）；流程简单到几个函数就能写完（用普通 Python，别上图）。

上一篇：07 Router 路由 ｜ 回目录：README ｜ 实战案例：12 用 Router 搭多源知识库（Router 本身就是一个 custom workflow）

08 Custom workflow 自定义工作流：流程由你画，该用代码的地方就用代码
https://aicooleo.cn/blog/posts/multiagent-08-custom-workflow/
作者
aicooleo
发布于
2026-08-28
许可协议
CC BY-NC-SA 4.0
多 Agent 精讲 · 第 8 / 22 篇 返回合集