LangGraph 子图架构实现 AI 工作流

01. 子图结构与应用场景

对于一些更加复杂的系统，子图是一个非常有用的设计原则。使用子图允许在图的不同部分创建和管理不同的状态，这样可以轻松利用 LangGraph 实现类似 多智能体 亦或者 AI工作流 之类的功能，每个功能之间相互独立隔离，最后组装成一个大型复杂应用。

大家如果有使用过 Dify 、 Coze 等 AI 应用开发平台的 工作流 ，亦或者是 MetaGPT 、 AutoGPT 、 BabyAGI 等智能体框架，在这些功能下就涉及到了 多智能体 亦或者 多Agent应用 的相互协作与组成。

前面我们说过，LangGraph 的节点可以是任意的 Python 函数或者是 Runnable可运行组件 ，并且 图程序 经过编译后就是一个 Runnable可运行组件 ，所以我们可以考虑将其中一个 图程序 作为另外一个 图程序 的节点，这样就变相在 LangGraph 中去实现子图，从而将一些功能相近的节点单独组装成图，单独进行状态的管理。

而创建 子图 最核心的部分要认识到 图 之间的信息传递， 入口图 是父级，两个子图都被定义成 入口图 的节点，并且两个子图都继承了父级 入口图 的状态，并且每个子图都可以拥有自己的私有状态，任何想传回父级 入口图 的值，只需要在入口图中定义即可。

另外需要注意的是，如果定义了 多边 ，那么必须定义 归纳函数 ，因为多边的执行是 并行的 ，如果不定义归纳函数，数据会直接覆盖，由于并行的顺序是不确定的，如果不定义 归纳函数 ，数据可能会出现相互覆盖的问题，实际上在 LangGraph 编译的过程中，如果一个图有多边并行的情况，并且没有为每个字段都定义 归纳函数 ，会直接抛出错误。

02. LangGraph 实现示例

例如我们实现一个 营销智能体 ，其功能为 根据用户传递的原始问题生成一篇【直播带货】脚本，一篇【小红书推广】文案 ，在这里用户传递一段原始 Prompt，会调用两个 Agent 智能体并行完成各自的任务，最后再进行合并输出。

针对这类需求，我们可以使用 单图结构 来构建，也可以创建 多图结构 来构建，更推荐使用 多图结构 ，其优势也非常明显：

• 多图结构状态设计更简单，不用一次性考虑所有 Agent 智能体的状态，可以每个智能体单独管理自己的状态。

• 多图结构更易于扩展，后续需要添加多一个 百度SEO推广文案Agent ，只需添加多一个节点即可，程序无需大量调整。

实现示例如下：

from typing import TypedDict, Any, Annotated

import dotenv
from langchain_community.tools import GoogleSerperRun
from langchain_community.utilities import GoogleSerperAPIWrapper
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.runnables import RunnableConfig
from langchain_openai import ChatOpenAI
from langgraph.graph import MessagesState, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition

dotenv.load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini")


class GoogleSerperArgsSchema(BaseModel):
    query: str = Field(description="执行谷歌搜索的查询语句")


google_serper = GoogleSerperRun(
    api_wrapper=GoogleSerperAPIWrapper(),
    args_schema=GoogleSerperArgsSchema,
)


def reduce_str(left: str | None, right: str | None) -> str:
    if right is not None and right != "":
        return right
    return left


class AgentState(TypedDict):
    query: Annotated[str, reduce_str]  
    live_content: Annotated[str, reduce_str]  
    xhs_content: Annotated[str, reduce_str]  


class LiveAgentState(AgentState, MessagesState):
    """直播文案智能体状态"""
    pass


class XHSAgentState(AgentState):
    """小红书文案智能体状态"""
    pass


def chatbot_live(state: LiveAgentState, config: RunnableConfig) -> Any:
    """直播文案智能体聊天机器人节点"""
    
    prompt = ChatPromptTemplate.from_messages([
        (
            "system",
            "你是一个拥有10年经验的直播文案专家，请根据用户提供的产品整理一篇直播带货脚本文案，如果在你的知识库内找不到关于该产品的信息，可以使用搜索工具。"
        ),
        ("human", "{query}"),
        ("placeholder", "{chat_history}")
    ])
    chain = prompt | llm.bind_tools([google_serper])

    
    ai_message = chain.invoke({"query": state["query"], "chat_history": state["messages"]})
    
    return {
        "messages": [ai_message],
        "live_content": ai_message.content,
    }



live_agent_graph = StateGraph(LiveAgentState)

live_agent_graph.add_node("chatbot_live", chatbot_live)
live_agent_graph.add_node("tools", ToolNode([google_serper]))

live_agent_graph.set_entry_point("chatbot_live")
live_agent_graph.add_conditional_edges("chatbot_live", tools_condition)
live_agent_graph.add_edge("tools", "chatbot_live")


def chatbot_xhs(state: XHSAgentState, config: RunnableConfig) -> Any:
    """小红书文案智能体聊天节点"""
    
    prompt = ChatPromptTemplate.from_messages([
        ("system",
         "你是一个小红书文案大师，请根据用户传递的商品名，生成一篇关于该商品的小红书笔记文案，注意风格活泼，多使用emoji表情。"),
        ("human", "{query}"),
    ])
    chain = prompt | llm | StrOutputParser()

    
    return {"xhs_content": chain.invoke({"query": state["query"]})}



xhs_agent_graph = StateGraph(XHSAgentState)

xhs_agent_graph.add_node("chatbot_xhs", chatbot_xhs)

xhs_agent_graph.set_entry_point("chatbot_xhs")
xhs_agent_graph.set_finish_point("chatbot_xhs")



def parallel_node(state: AgentState, config: RunnableConfig) -> Any:
    return state


agent_graph = StateGraph(AgentState)
agent_graph.add_node("parallel_node", parallel_node)
agent_graph.add_node("live_agent", live_agent_graph.compile())
agent_graph.add_node("xhs_agent", xhs_agent_graph.compile())

agent_graph.set_entry_point("parallel_node")
agent_graph.add_edge("parallel_node", "live_agent")
agent_graph.add_edge("parallel_node", "xhs_agent")

agent_graph.set_finish_point("live_agent")
agent_graph.set_finish_point("xhs_agent")


agent = agent_graph.compile()


print(agent.invoke({"query": "潮汕牛肉丸"}))

输出内容：

{'query': '潮汕牛肉丸', 'live_content': '### 潮汕牛肉丸直播带货脚本\n\n---\n\n**开场白（约30秒）**\n- 主持人：大家好，欢迎来到我们的直播间！今天我要给大家介绍的是来自潮汕地区的经典美食——潮汕牛肉丸！它可不是普通的牛肉丸，而是有着百年历史的传统美食，Q弹爽口，绝对让你一试成主顾！\n\n---\n\n**产品介绍（约1分钟）**\n- 主持人：潮汕牛肉丸分为牛肉丸和牛筋丸两种，牛肉丸肉质细嫩，口感滑嫩；而牛筋丸则在牛肉中加入了牛筋，增加了嚼劲，尤其适合喜欢丰富口感的朋友们！每一口都能感受到牛肉的鲜香与弹性，真的是吃了让人难以忘怀的美味。\n\n---\n\n**制作过程（约1分钟）**\n- 主持人：你们知道吗？潮汕牛肉丸的制作过程可是相当讲究哦！首先选用新鲜的牛后腿肉，剁成肉泥后，加入适量的肥肉，这样能让牛肉丸更加鲜美。然后再加入盐、胡椒粉等调味料，搅拌均匀，最后手打成型，冷冻后再煮熟，确保每一颗牛肉丸都弹牙十足！\n\n---\n\n**食用推荐（约1分钟）**\n- 主持人：潮汕牛肉丸的吃法多种多样，可以搭配牛肉丸粉丝汤、火锅，或者直接煮熟后沾上沙茶酱，都是极好的选择！无论是家庭聚餐还是朋友聚会，潮汕牛肉丸都能为你的餐桌增添不少色彩。\n\n---\n\n**优惠活动（约1分钟）**\n- 主持人：现在下单还有特别优惠哦！前50名下单的朋友将享受买一送一的超值活动！还有机会获得我们精美的潮汕牛肉丸专属礼盒，绝对是送礼自用两相宜的好选择！\n\n---\n\n**互动环节（约1分钟）**\n- 主持人：大家有没有吃过潮汕牛肉丸呢？欢迎在评论区分享你的体验和吃法！同时，如果你对潮汕牛肉丸有任何问题，也可以随时问我哦！\n\n---\n\n**结束语（约30秒）**\n- 主持人：感谢大家今天的观看，潮汕牛肉丸就等你来尝鲜！记得关注我们的直播间，获取更多美食资讯和优惠！我们下次见，祝大家生活愉快，吃得开心！\n\n---\n\n**背景音乐与视觉特效**\n- 在整个直播过程中，适当穿插一些美食的诱人画面，以及轻松愉快的背景音乐，提升观众的观看体验和购买欲望。\n\n---\n\n希望这个脚本能够帮助你成功地进行潮汕牛肉丸的直播带货！如果需要调整或添加内容，随时告诉我！', 'xhs_content': '✨🍜【潮汕牛肉丸，舌尖上的幸福】🍜✨\n\n小伙伴们，今天要跟大家分享一款让我欲罢不能的美食——潮汕牛肉丸！😋❤️\n\n🔍 外观：看这颗颗圆润的牛肉丸，真的是太诱人了！光滑的表面，隐隐透着肉汁的光泽，简直让人垂涎欲滴！🍽️\n\n🥩 口感：一口咬下去，外皮Q弹，内里鲜嫩多汁，真的是肉汁四溅，满口幸福感！每一口都仿佛在跟味蕾跳舞，太过瘾了！💃✨\n\n🍲 推荐吃法：直接放入热汤中煮，或者搭配一碗米粉，绝对是绝配！还可以加点香菜和辣椒，瞬间提升风味！🌶️🥬\n\n📷 拍照打卡：记得拍下这美味的瞬间，分享给你的朋友们，让他们也感受这份美好！📸💕\n\n👩\u200d🍳 DIY小技巧：如果你想在家尝试制作，记得选用新鲜的牛肉，搅拌到起胶，再加入一些淀粉，牛肉丸的口感会更加Q弹哦！✨\n\n总之，潮汕牛肉丸绝对是我心目中的人间美味！🍖💖 你们吃过吗？快来评论区分享你们的体验吧！👇👇\n\n#潮汕牛肉丸 #美食分享 #舌尖上的中国 #吃货日记 #幸福感满满'}

最新版 LangChain 用法提示

• 新版 LangChain 更推荐用 LCEL、Runnable、ChatModel.bind_tools()、结构化输出和 LangGraph 来组织复杂链路；老式 Chain、部分 AgentExecutor 写法可以读懂，但新项目应优先选择更清晰的图或 Runnable 编排。

• 工具调用相关代码要区分两层：模型是否原生支持 tool/function calling，以及业务侧如何定义工具 schema、参数校验、错误兜底和观测日志。

• 如果示例中的导入路径和你当前安装版本不同，优先查当前版本包内导出位置；常见迁移方向是从 langchain 拆到 langchain-core、langchain-community、langchain-openai、langgraph 等包。

拓展

• 工具或插件不要只看能不能调通，更要看是否可观测、可限流、可重试、可审计。联网类工具还要处理超时、空结果、搜索噪声和结果时效性。

• Agent 场景里，Prompt 只是调度策略的一部分；工具描述、参数 schema、历史状态、错误反馈、停止条件和人工介入点同样会影响最终稳定性。

常见问题

• 为什么模型没有调用工具？常见原因是工具描述不清晰、参数 schema 过宽或过窄、用户问题不需要工具、模型本身不支持工具调用，或者工具绑定位置不对。

• 为什么工具调用后回答仍然不准？先看工具返回是否正确，再看工具结果是否被放回模型上下文，最后检查输出解析、历史消息和异常兜底是否覆盖了真实错误。

面试题

• 解释函数调用、工具调用和 Agent 的区别。

• LangChain 中 tool schema 的作用是什么？为什么参数校验对生产环境很重要？

• ReACT Agent 和 tool-calling Agent 的核心差异是什么？分别适合什么场景？

• LangGraph 相比 LCEL 更适合解决哪些复杂编排问题？

生产问题排查

问题	常见原因	处理方式
工具没有被调用	工具描述弱、绑定失败、模型不支持	打印绑定后的模型配置，补充工具描述，换用支持工具调用的模型
参数格式错误	schema 设计不清晰，模型生成字段不稳定	使用 Pydantic/JSON Schema 校验，失败后把错误反馈给模型重试
联网结果不可用	搜索为空、接口超时、命中低质量页面	增加超时、重试、结果过滤、来源白名单和降级回答
Agent 循环不停止	缺少终止条件或工具返回被误判	设置最大迭代次数，记录每轮 thought/action/observation，增加停止规则
线上难以复现	缺少输入、工具请求和模型响应日志	给每次调用加 trace id，记录工具入参、出参、耗时和异常