Langgraph Studio分层多智能体实现求助：示例与结构指导

核心结构思路 #

分层智能体的核心是上层调度智能体负责任务拆解与分配，下层执行智能体负责具体任务处理，在Langgraph Studio中需要通过标准化状态结构、定义节点逻辑和配置路由关联来实现协作。

代码片段示例 #

1. 定义下层执行智能体 #

先创建两个专注具体任务的执行智能体，分别处理数据分析和文本生成：

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

# 数据分析执行智能体
data_analyst_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是专业数据分析师，擅长处理数据统计、趋势分析任务，输出简洁专业的分析结论。"),
    ("user", "{input}")
])
data_analyst_agent = create_react_agent(
    ChatOpenAI(model="gpt-4o"),
    tools=[],  # 可按需添加数据查询、计算类工具
    prompt=data_analyst_prompt
)

# 文本生成执行智能体
content_writer_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是资深文案，擅长把专业内容转化为通俗易懂的运营文案，风格要贴合业务场景。"),
    ("user", "{input}")
])
content_writer_agent = create_react_agent(
    ChatOpenAI(model="gpt-4o"),
    tools=[],  # 可按需添加素材检索、排版类工具
    prompt=content_writer_prompt
)

2. 定义上层调度智能体与状态结构 #

调度智能体负责判断任务类型、分配执行节点，同时需要统一的状态结构来传递任务数据：

from langgraph.graph import StateGraph, END
from typing import TypedDict, List

# 全局共享状态结构（Langgraph Studio协作的核心）
class AgentState(TypedDict):
    task: str
    intermediate_results: List[str]
    final_result: str

# 调度逻辑：判断任务类型并分配
def router_agent(state: AgentState):
    task = state["task"]
    if any(keyword in task for keyword in ["数据", "分析", "统计", "指标"]):
        # 调用数据分析智能体
        analysis_result = data_analyst_agent.invoke({"input": task})
        state["intermediate_results"].append(analysis_result["output"])
        # 若任务需要生成报告，转至文案智能体
        if any(keyword in task for keyword in ["报告", "文案", "解读"]):
            return "to_writer"
        state["final_result"] = analysis_result["output"]
        return END
    elif any(keyword in task for keyword in ["文案", "写作", "生成", "解读"]):
        writer_result = content_writer_agent.invoke({"input": task})
        state["final_result"] = writer_result["output"]
        return END
    else:
        state["final_result"] = "无法识别任务类型，请明确需求方向。"
        return END

# 文案智能体的后续处理逻辑
def writer_process(state: AgentState):
    input_text = "基于以下分析结果生成业务报告：\n" + "\n".join(state["intermediate_results"])
    writer_result = content_writer_agent.invoke({"input": input_text})
    state["final_result"] = writer_result["output"]
    return END

3. 构建Langgraph Studio兼容的图谱 #

将智能体和逻辑组装成可导入Studio的图谱结构：

# 创建状态图谱
graph = StateGraph(AgentState)

# 添加节点
graph.add_node("router", router_agent)
graph.add_node("writer", writer_process)

# 配置路由关联（需与Studio可视化界面的连线一致）
graph.set_entry_point("router")
graph.add_edge("router", "writer")
graph.add_edge("router", END)
graph.add_edge("writer", END)

# 编译图谱（可导出为Langgraph Studio支持的格式）
app = graph.compile()

# 测试运行
test_state = {
    "task": "分析近半年用户留存数据，生成一份面向运营团队的解读报告",
    "intermediate_results": [],
    "final_result": ""
}
result = app.invoke(test_state)
print(result["final_result"])

Langgraph Studio配置关键注意事项 #

• 状态结构必须统一：所有智能体节点必须使用同一个TypedDict状态类，否则Studio无法在节点间传递数据。
• 节点与路由可视化对齐：在Studio的可视化界面中，节点名称要和代码中add_node的命名一致，拖拽连线要匹配代码里的add_edge逻辑。
• 工具配置同步：如果执行智能体需要调用工具，需先在Studio的"Tools"面板配置工具，再在代码中引用工具ID，避免工具调用失败。
• 调试断点排查：开启Studio的调试模式，查看每个节点的输入输出数据，快速定位分层协作中的问题（比如调度智能体是否正确分配任务）。

内容的提问来源于stack exchange，提问作者Melvin Dow