┌──────────┐
         │ Supervisor│
         │  Agent   │
    ┌────┴──┬───┬───┴────┐
    ▼       ▼   ▼        ▼
┌──────┐┌──────┐┌──────┐┌──────┐
│Research││ Code ││ Data ││ Writer│
│ Agent ││Agent ││Agent ││ Agent │
└──────┘└──────┘└──────┘└──────┘

from langgraph.supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent

# 底层研究团队
research_agent = create_react_agent(model, tools=[search_tool])
doc_agent = create_react_agent(model, tools=[doc_tool])
research_team = create_supervisor(
    [research_agent, doc_agent],
    model=model,
    prompt="你是研究团队的管理者..."
)

# 底层写作团队
writer_agent = create_react_agent(model, tools=[write_tool])
review_agent = create_react_agent(model, tools=[review_tool])
writing_team = create_supervisor(
    [writer_agent, review_agent],
    model=model,
    prompt="你是写作团队的管理者..."
)

# 顶层Supervisor
top_supervisor = create_supervisor(
    [research_team, writing_team],
    model=model,
    prompt="你是总管理者，协调研究团队和写作团队..."
)

{
  "@type": "AgentCard",
  "name": "财务报表分析Agent",
  "description": "分析企业财务报表并生成洞察报告",
  "url": "https://agents.example.com/finance-analyzer",
  "capabilities": {
    "streaming": true,
    "pushNotifications": true
  },
  "skills": [
    {
      "name": "balance-sheet-analysis",
      "description": "分析资产负债表"
    },
    {
      "name": "cashflow-forecast",
      "description": "现金流量预测"
    }
  ],
  "authentication": {
    "schemes": ["OAuth2"]
  }
}

from agents import Agent, Runner, handoff, RunContext

# 定义专业Agent
triage_agent = Agent(
    name="分流Agent",
    instructions="分析用户请求，决定转交给哪个专业Agent",
    handoffs=[
        handoff(agent=research_agent, 
                description="处理研究和信息查询请求"),
        handoff(agent=coding_agent,
                description="处理编程和代码相关请求"),
        handoff(agent=writing_agent,
                description="处理内容创作请求"),
    ]
)

# Handoff核心：LLM自主决策交接
result = await Runner.run(triage_agent, messages=conversation)

from agents import GuardrailFunctionOutput, Agent, InputGuardrail

def check_financial_advice(context, agent, input_data):
    """防止Agent提供未授权的财务建议"""
    if "投资建议" in input_data and not context.get("licensed"):
        return GuardrailFunctionOutput(
            tripwire_triggered=True,
            output="未持牌，不可提供投资建议"
        )
    return GuardrailFunctionOutput(tripwire_triggered=False)

agent = Agent(
    name="FinancialAgent",
    input_guardrails=[InputGuardrail(guardrail_function=check_financial_advice)]
)

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main():
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    analyst = AssistantAgent(
        name="数据分析师",
        model_client=model_client,
        system_message="你负责分析数据并提取洞察"
    )
    writer = AssistantAgent(
        name="报告撰写员",
        model_client=model_client,
        system_message="你根据分析结果撰写报告"
    )

    # RoundRobin团队：轮流发言
    team = RoundRobinGroupChat(
        participants=[analyst, writer],
        max_turns=6
    )

    result = await team.run(task="分析Q1销售数据并撰写总结报告")

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    current_agent: str
    task_complete: bool

def supervisor_node(state: AgentState) -> AgentState:
    """Supervisor决策节点"""
    last_msg = state["messages"][-1]
    # LLM决策下一步
    next_agent = llm_decide(last_msg)
    return {"current_agent": next_agent}

def researcher_node(state: AgentState) -> AgentState:
    """研究Agent节点"""
    result = research_tool(state["messages"][-1])
    return {"messages": [result]}

# 构建图
graph = StateGraph(AgentState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("researcher", researcher_node)
graph.add_node("writer", writer_node)

# 定义边（条件路由）
graph.add_conditional_edges(
    "supervisor",
    lambda s: s["current_agent"],
    {"researcher": "researcher", "writer": "writer", "END": END}
)
graph.add_edge("researcher", "supervisor")
graph.add_edge("writer", "supervisor")

app = graph.compile()

用户需求 → 产品经理(PRD) → 架构师(系统设计) → 
项目经理(任务分解) → 工程师(代码实现) → QA(测试)

from camel.agents import ChatAgent
from camel.messages import BaseMessage
from camel.societies import RolePlaying

# 定义角色扮演会话
role_play = RolePlaying(
    assistant_role_name="Python专家",
    user_role_name="初学者",
    assistant_agent_kwargs={"model": "gpt-4o"},
    user_agent_kwargs={"model": "gpt-4o"},
    task_prompt="教会初学者理解Python装饰器",
    with_task_specify=True
)

# 自动运行角色扮演对话
chat_turns = 5
for _ in range(chat_turns):
    assistant_response = role_play.assistant_agent.step(
        role_play.user_agent.msgs[-1]
    )
    user_response = role_play.user_agent.step(
        assistant_response.msgs[-1]
    )

阶段1: 单Agent + 工具 → 验证业务价值
阶段2: Supervisor + 2-3个Worker Agent → 处理多领域任务
阶段3: 层级式团队 + 外部Agent集成 → 企业级编排
阶段4: 混合编排 + A2A/MCP协议栈 → 跨组织互操作

LangSmith / Arize Phoenix → Trace追踪
Prometheus + Grafana → 性能监控
自定义Dashboard → 业务指标追踪

"""
三层多智能体研究系统
Layer 1: 顶级Supervisor（任务分类）
Layer 2: 团队Supervisor（领域协调）
Layer 3: 专项Worker（执行具体任务）
"""
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated, Literal
import operator

# === 状态定义 ===
class ResearchState(TypedDict):
    query: str
    domain: Literal["tech", "market", "academic"]
    subtasks: list[str]
    findings: Annotated[list[str], operator.add]
    report: str
    current_step: str

# === Layer 3: Worker Agents ===
def web_searcher(state: ResearchState) -> dict:
    """网络搜索Worker"""
    task = state["subtasks"].pop(0) if state["subtasks"] else state["query"]
    results = search_tool(task)  # 假设已定义
    return {"findings": [f"[Web] {results}"], "current_step": "web_done"}

def data_analyst(state: ResearchState) -> dict:
    """数据分析Worker"""
    data = extract_data(state["findings"])
    analysis = analyze_with_llm(data)
    return {"findings": [f"[Analysis] {analysis}"], "current_step": "analysis_done"}

def report_writer(state: ResearchState) -> dict:
    """报告撰写Worker"""
    report = synthesize_report(state["findings"])
    return {"report": report, "current_step": "complete"}

# === Layer 2: 团队Supervisor ===
def research_supervisor(state: ResearchState) -> dict:
    """研究团队协调者"""
    if not state.get("findings"):
        return {"current_step": "search", "subtasks": decompose(state["query"])}
    elif len(state["findings"]) < 3:
        return {"current_step": "analyze"}
    else:
        return {"current_step": "write"}

# === Layer 1: 顶级路由 ===
def top_router(state: ResearchState) -> dict:
    """顶级路由：判断请求领域"""
    domain = classify_domain(state["query"])
    return {"domain": domain, "current_step": "delegate"}

# === 构建图 ===
graph = StateGraph(ResearchState)

# 添加节点
graph.add_node("router", top_router)
graph.add_node("team_supervisor", research_supervisor)
graph.add_node("searcher", web_searcher)
graph.add_node("analyst", data_analyst)
graph.add_node("writer", report_writer)

# 定义路由边
graph.add_edge(START, "router")
graph.add_edge("router", "team_supervisor")
graph.add_conditional_edges(
    "team_supervisor",
    lambda s: s["current_step"],
    {"search": "searcher", "analyze": "analyst", 
     "write": "writer", "complete": END}
)
graph.add_edge("searcher", "team_supervisor")
graph.add_edge("analyst", "team_supervisor")
graph.add_edge("writer", END)

app = graph.compile()