01. 子图结构与应用场景

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

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

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

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

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

02. LangGraph 实现示例

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

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

1. 多图结构状态设计更简单，不用一次性考虑所有 Agent 智能体的状态，可以每个智能体单独管理自己的状态。
2. 多图结构更易于扩展，后续需要添加多一个 百度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:

    """直播文案智能体聊天机器人节点"""

    # 1.创建提示模板+链应用

    prompt = ChatPromptTemplate.from_messages([

        (

            "system",

            "你是一个拥有10年经验的直播文案专家，请根据用户提供的产品整理一篇直播带货脚本文案，如果在你的知识库内找不到关于该产品的信息，可以使用搜索工具。"

        ),

        ("human", "{query}"),

        ("placeholder", "{chat_history}")

    ])

    chain = prompt | llm.bind_tools([google_serper])

    # 2.调用链并生成ai消息

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

    

    return {

        "messages": [ai_message],

        "live_content": ai_message.content,

    }

# 1.创建子图1并添加节点、添加边

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:

    """小红书文案智能体聊天节点"""

    # 1.创建提示模板+链

    prompt = ChatPromptTemplate.from_messages([

        ("system",

         "你是一个小红书文案大师，请根据用户传递的商品名，生成一篇关于该商品的小红书笔记文案，注意风格活泼，多使用emoji表情。"),

        ("human", "{query}"),

    ])

    chain = prompt | llm | StrOutputParser()

    # 2.调用链并生成内容更新状态

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

# 2.创建子图2并添加节点、添加边

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

# 3.创建入口图并添加节点、边

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

# 4.编译入口图

agent = agent_graph.compile()

# 5.执行入口图并打印结果

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#潮汕牛肉丸 #美食分享 #舌尖上的中国 #吃货日记 #幸福感满满'}