---

name: multi-agent-architect description: "设计和优化基于 LangGraph、LangChain 和 DeepAgents 的生产级多智能体系统，用于复杂 AI 工作流。" risk: safe source: community metadata: category: ai-engineering source_repo: pravin-python/antigravity-awesome-skills source_type: community date_added: "2025-05-07" author: community tags: [langgraph, langchain, multi-agent, orchestration, deepagents, rag, tool-calling] tools: [claude, cursor, gemini] license: "MIT" license_source: "https://github.com/pravin-python/antigravity-awesome-skills/blob/main/LICENSE"

多智能体架构师与更新技能

概述

此技能将 Claude 转变为专精于 LangGraph、LangChain 和 DeepAgents 的高级 AI 多智能体架构师。它提供结构化工作流，用于创建和更新生产级多智能体系统——包括监督智能体、规划器、研究员、编码器和基于记忆的自主管道。当你需要设计、构建、调试或扩展任何多智能体 AI 系统时，请使用此技能。

如果此技能引用了外部 GitHub 仓库的材料，请同时声明：

• source_repo: owner/repo
• source_type: official 或 source_type: community

何时使用此技能

• 当你需要从零开始创建新的智能体或多智能体工作流时
• 当使用 LangGraph 状态图、节点、边或条件路由时
• 当用户询问智能体通信、记忆系统或工具调用管道时
• 当调试或优化现有 LangChain/LangGraph 智能体系统时
• 当设计监督者、规划器、研究、编码或验证智能体角色时
• 当集成具有层级规划和委派功能的 DeepAgents 时

工作原理

步骤 1：理解目标

在编写任何代码之前，先明确：

• 此智能体系统必须实现的业务目标是什么？
• 需要哪些智能体角色（监督者、规划器、研究员、编码器、验证器）？
• 每个智能体需要哪些工具？
• 需要什么记忆策略（Redis、向量数据库、LangChain Memory）？
• 什么通信协议连接各智能体（共享状态、消息传递）？

步骤 2：定义状态模式

所有智能体共享一个通过图传递的类型化状态对象：

from typing import TypedDict

class AgentState(TypedDict):
    user_goal: str
    tasks: list[str]
    completed_tasks: list[str]
    next_agent: str
    context: dict
    step_count: int          # guards against infinite loops
    error: str | None

步骤 3：定义智能体节点

每个智能体是一个异步函数，从状态读取并返回更新后的状态：

import logging
from langchain_openai import ChatOpenAI

logger = logging.getLogger(__name__)

async def research_node(state: AgentState) -> AgentState:
    logger.info("research_node: starting")
    llm = ChatOpenAI(model="gpt-4o")
    result = await llm.bind_tools(research_tools).ainvoke(state["user_goal"])
    state["context"]["research"] = result.content
    state["next_agent"] = "coder"
    return state

步骤 4：构建 LangGraph

使用边和条件路由将节点连接在一起：

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode

def build_graph() -> StateGraph:
    graph = StateGraph(AgentState)

    graph.add_node("supervisor", supervisor_node)
    graph.add_node("research",   research_node)
    graph.add_node("coder",      coding_node)
    graph.add_node("validator",  validation_node)
    graph.add_node("tools",      ToolNode(all_tools))

    graph.set_entry_point("supervisor")

    graph.add_conditional_edges(
        "supervisor",
        route_next,
        {"research": "research", "coder": "coder", "end": END}
    )

    graph.add_edge("research",  "supervisor")
    graph.add_edge("coder",     "validator")
    graph.add_edge("validator", "supervisor")

    return graph.compile()

def route_next(state: AgentState) -> str:
    if state["step_count"] > 20:
        return "end"
    return state["next_agent"]

步骤 5：添加记忆

from langchain_community.chat_message_histories import RedisChatMessageHistory

def get_memory(session_id: str):
    return RedisChatMessageHistory(
        session_id=session_id,
        url=os.getenv("REDIS_URL"),
        ttl=3600
    )

步骤 6：运行图

async def run(user_goal: str, session_id: str):
    graph = build_graph()
    initial_state = AgentState(
        user_goal=user_goal,
        tasks=[],
        completed_tasks=[],
        next_agent="supervisor",
        context={},
        step_count=0,
        error=None,
    )
    return await graph.ainvoke(initial_state)

步骤 7：通过 FastAPI 暴露（可选）

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class RunRequest(BaseModel):
    goal: str
    session_id: str

@app.post("/run")
async def run_agent(req: RunRequest)