自定义工作流

在自定义工作流架构中，你可以使用 LangGraph 定义自己的定制化执行流程。你可以完全控制图结构——包括顺序步骤、条件分支、循环和并行执行。

主要特点

• 完全控制图结构
• 混合确定性逻辑与智能体行为
• 支持顺序步骤、条件分支、循环和并行执行
• 可将其他模式作为节点嵌入工作流中

使用场景

当标准模式（子智能体、技能等）不符合你的需求时，当你需要混合确定性逻辑与智能体行为时，或者你的用例需要复杂路由或多阶段处理时，请使用自定义工作流。

工作流中的每个节点可以是一个简单函数、一个LLM调用，或是一个完整的带有工具的智能体。你也可以在自定义工作流中组合其他架构——例如，将多智能体系统作为单个节点嵌入。

有关自定义工作流的完整示例，请参阅下面的教程。

教程：构建具有路由功能的多源知识库

路由模式是自定义工作流的一个示例。本教程将逐步指导你构建一个并行查询GitHub、Notion和Slack，然后综合结果的路由器。

了解更多

基础实现

核心思路是，你可以在任何LangGraph节点中直接调用LangChain智能体，从而将自定义工作流的灵活性与预构建智能体的便利性结合起来：

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

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

def agent_node(state: State) -> dict:
    """一个调用LangChain智能体的LangGraph节点。"""
    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()
)

示例：RAG管道

一个常见的用例是将检索与智能体结合。这个示例构建了一个WNBA数据助手，它可以从知识库中检索信息，并能获取实时新闻。

自定义RAG工作流

该工作流展示了三种类型的节点：

• 模型节点（重写）：使用结构化输出重写用户查询以优化检索。
• 确定性节点（检索）：执行向量相似性搜索——不涉及LLM。
• 智能体节点（智能体）：基于检索到的上下文进行推理，并能通过工具获取额外信息。

你可以使用LangGraph状态在工作流步骤之间传递信息。这允许工作流的每个部分读取和更新结构化字段，从而轻松地在节点之间共享数据和上下文。

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([
    # 阵容
    "New York Liberty 2024 roster: Breanna Stewart, Sabrina Ionescu, Jonquel Jones, Courtney Vandersloot.",
    "Las Vegas Aces 2024 roster: A'ja Wilson, Kelsey Plum, Jackie Young, Chelsea Gray.",
    "Indiana Fever 2024 roster: Caitlin Clark, Aliyah Boston, Kelsey Mitchell, NaLyssa Smith.",
    # 比赛结果
    "2024 WNBA Finals: New York Liberty defeated Minnesota Lynx 3-2 to win the championship.",
    "June 15, 2024: Indiana Fever 85, Chicago Sky 79. Caitlin Clark had 23 points and 8 assists.",
    "August 20, 2024: Las Vegas Aces 92, Phoenix Mercury 84. A'ja Wilson scored 35 points.",
    # 球员数据
    "A'ja Wilson 2024 season stats: 26.9 PPG, 11.9 RPG, 2.6 BPG. Won MVP award.",
    "Caitlin Clark 2024 rookie stats: 19.2 PPG, 8.4 APG, 5.7 RPG. Won Rookie of the Year.",
    "Breanna Stewart 2024 stats: 20.4 PPG, 8.5 RPG, 3.5 APG.",
])
retriever = vector_store.as_retriever(search_kwargs={"k": 5})

@tool
def get_latest_news(query: str) -> str:
    """获取最新的WNBA新闻和更新。"""
    # 你的新闻API在这里
    return "Latest: The WNBA announced expanded playoff format for 2025..."

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

model = ChatOpenAI(model="gpt-4.1")

class RewrittenQuery(BaseModel):
    query: str

def rewrite_query(state: State) -> dict:
    """重写用户查询以优化检索。"""
    system_prompt = """重写此查询以检索相关的WNBA信息。
知识库包含：球队阵容、带比分的比赛结果和球员统计数据（PPG、RPG、APG）。
重点关注提到的具体球员姓名、球队名称或统计类别。"""
    response = model.with_structured_output(RewrittenQuery).invoke([
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": state["question"]}
    ])
    return {"rewritten_query": response.query}

def retrieve(state: State) -> dict:
    """基于重写后的查询检索文档。"""
    docs = retriever.invoke(state["rewritten_query"])
    return {"documents": [doc.page_content for doc in docs]}

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"])

---

Edit this page on GitHub or file an issue.

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.