使用者痛点一览

• **并行执行难**：传统方法只能顺序跑，无法同时生成多个文案。
• **状态共享麻烦**：各个智能体需要访问同一份查询、历史记录，手动同步容易出错。
• **工具调用混乱**：当需要外部搜索时LLM 与工具的交互往往导致循环调用或信息丢失。
• **流程可视化不足**：没有直观图表展示子图与主图的关系，难以快速定位问题。

LangGraph 子图概念速览

LangGraph 的子图可以像函数一样被复用。而且支持并行分发、状态归约和条件跳转. 通过把每个业务逻辑封装成子图，再在主图中并行触发，就能轻松解决上述痛点。

1️⃣ 初始化 LLM 与工具

import os
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.runnables import RunnableConfig
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph,add_messages
from langgraph.prebuilt import ToolNode。tools_condition
from pydantic.v1 import Field,BaseModel
dotenv.load_dotenv
# ==================== 初始化 LLM ====================
llm = ChatOpenAI(
model="qwen3-max",api_key=os.getenv,base_url=os.getenv
)
# ==================== 工具定义 ====================
class GoogleSerperArgsSchema:
query这方面，str = Field
google_serper = GoogleSerperRun(
api_wrapper=GoogleSerperAPIWrapper,args_schema=GoogleSerperArgsSchema,)

2️⃣ 状态归约函数 & 状态定义

# 字符串归约函数：保留非空新值，否则保留旧值
def reduce_str -> str:
if right is not None and right!= "":
return right
return left
# 主图状态定义，所有共享字段都放在这里
class AgentState:
query的观点是，Annotated # 原始问题/商品名
live_content: Annotated # 直播文案
xhs_content: Annotated # 文案
messages: Annotated,add_messages] # 对话历史
# 子图状态继承主图字段 + messages，用于类型检查更细粒度控制
class LiveAgentState: pass
class XHSAgentState: pass

3️⃣ 子图 – 直播文案智能体

def chatbot_live -> Any:
"""生成直播带货脚这篇文章案，可调用搜索工具"""
prompt = ChatPromptTemplate.from_messages()
chain = prompt | llm.bind_tools
ai_message = chain.invoke
return {
"messages":。"live_content": ai_message.content,}
# 建立 Live 子图
live_agent_graph = StateGraph
live_agent_graph.add_node
live_agent_graph.add_node)
live_agent_graph.set_entry_point
live_agent_graph.add_conditional_edges(
"chatbot_live",tools_condition # 若 LLM 决定调用工具则走 tools 节点，否则直接结束
)
live_agent_graph.add_edge # 工具返回后继续 LLM 推理
print.print_ascii)

⚡️ 小贴士：

* 将 `tools_condition` 放在入口后可让模型自动决定是否使用搜索； 如果不应该把直接跳到 END搜索，可。

4️⃣ 子图 – 笔记智能体

def chatbot_xhs -> Any:
"""生成风格活泼、带 emoji 的笔记"""
prompt = ChatPromptTemplate.from_messages()
chain = prompt | llm | StrOutputParser
return {"xhs_content": chain.invoke}
xhs_agent_graph = StateGraph
xhs_agent_graph.add_node
xhs_agent_graph.set_entry_point
xhs_agent_graph.set_finish_point
print.print_ascii)

⚡️ 小贴士：

* XHS 智能体不需要外部工具，流程更简单；如果未来想加入图片检索，只需再加一个 `ToolNode` 即可。

5️⃣ 主图 – 并行编排两个子图

def parallel_node -> Any:
"""把整个请求传递给两个子智能体"""
return state
agent_graph = StateGraph
agent_graph.add_node # 分发节点
agent_graph.add_node)
agent_graph.add_node)
agent_graph.set_entry_point
agent_graph.add_edge # 并行执行 Live 子图
agent_graph.add_edge # 并行执行 XHS 子图
# 两个子任务完成后退出主流程
agent_graph.set_finish_point
compiled_main = agent_graph.compile
print.print_ascii)

⚡️ 小贴士：

* 主图只做一次状态分发，无需重复计算；说起来，若想在主流里添加日志或监控，只需插入一个节点即可。

6️⃣ 执行示例 & 输出结果演示

result = compiled_main.invoke
print
print
print
print

✅ ****：通过将每个业务逻辑封装成 LangGraph 的子图。并在主 Graph 中并行触发，可解决掉“并行执行难”“状态同步麻烦”等痛点，实现高效、多模态的多智能体协作。