10d - Agent 通信模式

┌─────────┐    状态传递    ┌─────────┐    状态传递    ┌─────────┐
│ Agent A  │──────────────▶│ Agent B  │──────────────▶│ Agent C  │
│ (采集)   │               │ (分析)   │               │ (生成)   │
└─────────┘               └─────────┘               └─────────┘
     │                         │                         │
     ▼                         ▼                         ▼
┌─────────────────────────────────────────────────────────────┐
│                      共享状态存储                             │
│         (对话历史 / 中间结果 / 元数据)                        │
└─────────────────────────────────────────────────────────────┘

from langgraph.graph import StateGraph, END
from typing import TypedDict
 
class PipelineState(TypedDict):
    raw_data: str
    cleaned_data: str
    analysis: str
    report: str
 
def collect_agent(state: PipelineState) -> PipelineState:
    """Agent A：数据采集"""
    # 调用 LLM 或工具采集原始数据
    return {**state, "raw_data": "采集到的原始数据..."}
 
def clean_agent(state: PipelineState) -> PipelineState:
    """Agent B：数据清洗"""
    return {**state, "cleaned_data": f"清洗后: {state['raw_data']}"}
 
def analyze_agent(state: PipelineState) -> PipelineState:
    """Agent C：数据分析"""
    return {**state, "analysis": f"分析结果: {state['cleaned_data']}"}
 
def report_agent(state: PipelineState) -> PipelineState:
    """Agent D：报告生成"""
    return {**state, "report": f"最终报告: {state['analysis']}"}
 
# 构建顺序图
graph = StateGraph(PipelineState)
graph.add_node("collect", collect_agent)
graph.add_node("clean", clean_agent)
graph.add_node("analyze", analyze_agent)
graph.add_node("report", report_agent)
 
graph.set_entry_point("collect")
graph.add_edge("collect", "clean")
graph.add_edge("clean", "analyze")
graph.add_edge("analyze", "report")
graph.add_edge("report", END)
 
pipeline = graph.compile()
result = pipeline.invoke({"raw_data": "", "cleaned_data": "", "analysis": "", "report": ""})

                         ┌──────────────┐
                         │   分发器      │
                         │ (Dispatcher)  │
                         └──────┬───────┘
                    ┌───────────┼───────────┐
                    ▼           ▼           ▼
              ┌──────────┐┌──────────┐┌──────────┐
              │ Agent A  ││ Agent B  ││ Agent C  │
              │ (竞品)   ││ (用户)   ││ (趋势)   │
              └────┬─────┘└────┬─────┘└────┬─────┘
                   │           │           │
                   ▼           ▼           ▼
              ┌─────────────────────────────────┐
              │          聚合器                   │
              │     (Aggregator Agent)           │
              │   合并 + 去重 + 综合分析          │
              └─────────────────────────────────┘
                              │
                              ▼
                       ┌──────────────┐
                       │   最终报告    │
                       └──────────────┘

import asyncio
from agents import Agent, Runner
 
# 定义专业化子 Agent
competitor_agent = Agent(
    name="竞品分析师",
    instructions="你是竞品分析专家。分析给定行业的主要竞争对手，输出结构化报告。"
)
 
user_research_agent = Agent(
    name="用户研究员",
    instructions="你是用户研究专家。分析目标用户群体的需求、痛点和行为模式。"
)
 
trend_agent = Agent(
    name="趋势分析师",
    instructions="你是市场趋势专家。分析行业发展趋势和技术演进方向。"
)
 
# 聚合 Agent
synthesizer = Agent(
    name="综合分析师",
    instructions="""你是市场综合分析师。
    根据竞品分析、用户研究和趋势分析的结果，生成一份综合市场报告。
    要求：去除重复信息，识别交叉洞察，给出可执行建议。"""
)
 
async def fan_out_fan_in(query: str):
    """并行扇出 → 聚合扇入"""
    # 扇出：并行执行三个子任务
    results = await asyncio.gather(
        Runner.run(competitor_agent, query),
        Runner.run(user_research_agent, query),
        Runner.run(trend_agent, query),
    )
 
    # 扇入：聚合所有结果
    combined_input = f"""
    ## 竞品分析结果
    {results[0].final_output}
 
    ## 用户研究结果
    {results[1].final_output}
 
    ## 趋势分析结果
    {results[2].final_output}
 
    请综合以上三份报告，生成最终市场分析。
    """
    final = await Runner.run(synthesizer, combined_input)
    return final.final_output
 
# 执行
report = asyncio.run(fan_out_fan_in("分析 2026 年 AI 编码助手市场"))

                    ┌─────────────────────┐
                    │    总监 Agent        │
                    │  (Director/CEO)      │
                    │  任务分解 + 最终汇总  │
                    └──────────┬──────────┘
                 ┌─────────────┼─────────────┐
                 ▼             ▼             ▼
          ┌────────────┐┌────────────┐┌────────────┐
          │ 经理 A     ││ 经理 B     ││ 经理 C     │
          │ (前端组)   ││ (后端组)   ││ (测试组)   │
          └──────┬─────┘└──────┬─────┘└──────┬─────┘
            ┌────┴────┐   ┌────┴────┐   ┌────┴────┐
            ▼         ▼   ▼         ▼   ▼         ▼
        ┌──────┐ ┌──────┐┌──────┐┌──────┐┌──────┐┌──────┐
        │执行者││执行者 ││执行者││执行者 ││执行者││执行者 │
        │ UI   ││ 样式  ││ API  ││ DB   ││ 单元 ││ E2E  │
        └──────┘ └──────┘└──────┘└──────┘└──────┘└──────┘

        ─────────────────────────────────────────────────
        信息流：任务自上而下分解，结果自下而上汇报

from crewai import Agent, Task, Crew, Process
 
# 顶层：项目总监
director = Agent(
    role="项目总监",
    goal="协调前端、后端、测试团队完成产品开发",
    backstory="你是一位经验丰富的技术总监，擅长任务分解和团队协调。",
    allow_delegation=True
)
 
# 中层：团队经理
frontend_lead = Agent(
    role="前端负责人",
    goal="负责所有前端开发任务的分配和质量把控",
    backstory="你是资深前端架构师，精通 React 和设计系统。",
    allow_delegation=True
)
 
backend_lead = Agent(
    role="后端负责人",
    goal="负责 API 设计、数据库和服务端逻辑",
    backstory="你是后端架构师，精通微服务和数据库优化。",
    allow_delegation=True
)
 
qa_lead = Agent(
    role="测试负责人",
    goal="制定测试策略并确保产品质量",
    backstory="你是 QA 专家，精通自动化测试和质量保证。",
    allow_delegation=True
)
 
# 定义层级任务
planning_task = Task(
    description="分析产品需求，制定开发计划，分配给各团队负责人",
    agent=director,
    expected_output="包含前端、后端、测试三个模块的详细开发计划"
)
 
frontend_task = Task(
    description="根据开发计划完成前端组件开发",
    agent=frontend_lead,
    expected_output="前端组件代码和文档"
)
 
backend_task = Task(
    description="根据开发计划完成 API 和数据库开发",
    agent=backend_lead,
    expected_output="API 接口和数据库 schema"
)
 
qa_task = Task(
    description="根据开发计划制定并执行测试方案",
    agent=qa_lead,
    expected_output="测试报告和质量评估"
)
 
# 使用层级流程
crew = Crew(
    agents=[director, frontend_lead, backend_lead, qa_lead],
    tasks=[planning_task, frontend_task, backend_task, qa_task],
    process=Process.hierarchical,  # 关键：启用层级委托
    manager_agent=director
)
 
result = crew.kickoff()

                    ┌──────────────────┐
                    │   同一问题输入    │
                    └────────┬─────────┘
               ┌─────────────┼─────────────┐
               ▼             ▼             ▼
        ┌────────────┐┌────────────┐┌────────────┐
        │ Agent A    ││ Agent B    ││ Agent C    │
        │ (视角 1)   ││ (视角 2)   ││ (视角 3)   │
        │            ││            ││            │
        │ 独立推理   ││ 独立推理   ││ 独立推理   │
        │ 生成答案   ││ 生成答案   ││ 生成答案   │
        └──────┬─────┘└──────┬─────┘└──────┬─────┘
               │             │             │
               ▼             ▼             ▼
        ┌─────────────────────────────────────────┐
        │            投票 / 评分机制                │
        │                                         │
        │  策略 1: 多数投票 (Majority Vote)        │
        │  策略 2: 加权投票 (Weighted Vote)        │
        │  策略 3: 排名聚合 (Rank Aggregation)     │
        │  策略 4: 裁判 Agent (Judge Agent)        │
        └─────────────────────┬───────────────────┘
                              ▼
                    ┌──────────────────┐
                    │   最终共识结果    │
                    └──────────────────┘

import asyncio
from dataclasses import dataclass
from collections import Counter
 
@dataclass
class VoteResult:
    agent_name: str
    answer: str
    confidence: float
    reasoning: str
 
async def agent_vote(agent_name: str, prompt: str, perspective: str) -> VoteResult:
    """单个 Agent 独立投票"""
    system_prompt = f"你是{perspective}。请独立分析以下问题，给出你的判断。"
    # 调用 LLM API
    response = await call_llm(system_prompt, prompt)
    return VoteResult(
        agent_name=agent_name,
        answer=response["answer"],
        confidence=response["confidence"],
        reasoning=response["reasoning"]
    )
 
async def consensus_vote(question: str) -> dict:
    """共识投票主流程"""
    # 扇出：多个 Agent 独立投票
    votes = await asyncio.gather(
        agent_vote("安全专家", question, "资深安全工程师，关注漏洞和攻击面"),
        agent_vote("架构师", question, "系统架构师，关注可扩展性和性能"),
        agent_vote("业务分析师", question, "业务分析师，关注用户需求和商业价值"),
    )
 
    # 策略 1：简单多数投票
    answer_counts = Counter(v.answer for v in votes)
    majority_answer = answer_counts.most_common(1)[0][0]
 
    # 策略 2：加权投票（按置信度加权）
    weighted_scores = {}
    for v in votes:
        weighted_scores[v.answer] = weighted_scores.get(v.answer, 0) + v.confidence
    weighted_answer = max(weighted_scores, key=weighted_scores.get)
 
    return {
        "majority_vote": majority_answer,
        "weighted_vote": weighted_answer,
        "individual_votes": [
            {"agent": v.agent_name, "answer": v.answer,
             "confidence": v.confidence, "reasoning": v.reasoning}
            for v in votes
        ],
        "consensus_reached": len(set(v.answer for v in votes)) == 1
    }

                    ┌──────────────────┐
                    │   辩题 / 问题     │
                    └────────┬─────────┘
                             │
                    ┌────────▼─────────┐
                    │   主持人 Agent    │
                    │  (Moderator)     │
                    │  设定规则+判定    │
                    └────────┬─────────┘
                             │
              ┌──────────────┼──────────────┐
              ▼                             ▼
       ┌─────────────┐              ┌─────────────┐
       │  正方 Agent  │◄────辩论────▶│  反方 Agent  │
       │ (Proponent)  │   第 1 轮    │ (Opponent)   │
       └──────┬──────┘              └──────┬──────┘
              │                            │
              ▼                            ▼
       ┌─────────────┐              ┌─────────────┐
       │  反思 & 修正 │              │  反思 & 修正 │
       └──────┬──────┘              └──────┬──────┘
              │                            │
              ▼                            ▼
       ┌─────────────┐              ┌─────────────┐
       │  正方 Agent  │◄────辩论────▶│  反方 Agent  │
       │  (更新论点)  │   第 2 轮    │  (更新论点)  │
       └──────┬──────┘              └──────┬──────┘
              │                            │
              └──────────────┬─────────────┘
                             ▼
                    ┌──────────────────┐
                    │   裁判 Agent     │
                    │  综合评估 + 裁决  │
                    └──────────────────┘
                             │
                             ▼
                    ┌──────────────────┐
                    │   最终结论       │
                    │  (附完整辩论记录) │
                    └──────────────────┘

from langgraph.graph import StateGraph, END
from typing import TypedDict, List
 
class DebateState(TypedDict):
    topic: str
    round: int
    max_rounds: int
    proponent_args: List[str]
    opponent_args: List[str]
    moderator_notes: List[str]
    final_verdict: str
 
def proponent_argue(state: DebateState) -> DebateState:
    """正方 Agent：提出或强化论点"""
    context = f"""
    辩题：{state['topic']}
    当前第 {state['round']} 轮。
    反方上轮论点：{state['opponent_args'][-1] if state['opponent_args'] else '无'}
    
    请提出你的论点，回应反方质疑，强化你的立场。
    """
    # 调用 LLM
    argument = call_llm("你是正方辩手，坚定支持该方案。", context)
    return {
        **state,
        "proponent_args": state["proponent_args"] + [argument]
    }
 
def opponent_argue(state: DebateState) -> DebateState:
    """反方 Agent：质疑和反驳"""
    context = f"""
    辩题：{state['topic']}
    当前第 {state['round']} 轮。
    正方本轮论点：{state['proponent_args'][-1]}
    
    请质疑正方论点中的薄弱环节，提出反面论据。
    """
    argument = call_llm("你是反方辩手，坚定反对该方案。寻找漏洞和风险。", context)
    return {
        **state,
        "opponent_args": state["opponent_args"] + [argument]
    }
 
def moderator_evaluate(state: DebateState) -> DebateState:
    """主持人：评估是否继续辩论"""
    return {**state, "round": state["round"] + 1}
 
def should_continue(state: DebateState) -> str:
    """判断是否继续辩论"""
    if state["round"] >= state["max_rounds"]:
        return "judge"
    return "proponent"
 
def judge_verdict(state: DebateState) -> DebateState:
    """裁判 Agent：综合评估并裁决"""
    all_arguments = ""
    for i in range(len(state["proponent_args"])):
        all_arguments += f"\n--- 第 {i+1} 轮 ---\n"
        all_arguments += f"正方：{state['proponent_args'][i]}\n"
        if i < len(state["opponent_args"]):
            all_arguments += f"反方：{state['opponent_args'][i]}\n"
 
    verdict = call_llm(
        "你是公正的裁判。综合正反双方论点，给出最终裁决和建议。",
        f"辩题：{state['topic']}\n\n辩论记录：{all_arguments}"
    )
    return {**state, "final_verdict": verdict}
 
# 构建辩论图
graph = StateGraph(DebateState)
graph.add_node("proponent", proponent_argue)
graph.add_node("opponent", opponent_argue)
graph.add_node("moderator", moderator_evaluate)
graph.add_node("judge", judge_verdict)
 
graph.set_entry_point("proponent")
graph.add_edge("proponent", "opponent")
graph.add_edge("opponent", "moderator")
graph.add_conditional_edges("moderator", should_continue)
graph.add_edge("judge", END)
 
debate = graph.compile()
result = debate.invoke({
    "topic": "是否应该将单体应用拆分为微服务架构？",
    "round": 0, "max_rounds": 3,
    "proponent_args": [], "opponent_args": [],
    "moderator_notes": [], "final_verdict": ""
})

你的任务是什么类型？
│
├─ 步骤有严格先后依赖 ──────────────▶ 顺序交接
│
├─ 子任务相互独立 ──────────────────▶ 扇出/扇入
│
├─ 任务需要多层分解 ────────────────▶ 层级委托
│
├─ 需要高可靠性/降低幻觉 ──────────▶ 共识投票
│
└─ 需要深度分析正反面 ─────────────▶ 辩论/对抗

提示：实际项目中常常组合使用多种模式。
例如：层级委托 + 扇出/扇入（总监分解任务，各组并行执行）
     顺序交接 + 共识投票（流水线中关键步骤加入投票验证）

需求：评审一个"是否迁移到微服务架构"的技术方案

┌─────────────────────────────────────────────────────────┐
│ 第 1 阶段：并行调研（扇出/扇入）                          │
│                                                         │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐  │
│  │成本分析   │ │性能评估   │ │团队评估   │ │风险分析   │  │
│  │Agent     │ │Agent     │ │Agent     │ │Agent     │  │
│  └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬─────┘  │
│       └─────────────┴────────────┴─────────────┘       │
│                         │ 汇总                          │
└─────────────────────────┼───────────────────────────────┘
                          ▼
┌─────────────────────────────────────────────────────────┐
│ 第 2 阶段：对抗辩论（辩论/对抗）                          │
│                                                         │
│  正方（支持迁移）◄──── 3 轮辩论 ────▶ 反方（反对迁移）   │
│       基于调研数据展开论证                                │
└─────────────────────────┼───────────────────────────────┘
                          ▼
┌─────────────────────────────────────────────────────────┐
│ 第 3 阶段：专家投票（共识投票）                            │
│                                                         │
│  架构师 Agent + 运维 Agent + 业务 Agent                  │
│  各自基于辩论记录独立投票 → 加权汇总                      │
└─────────────────────────┼───────────────────────────────┘
                          ▼
                 ┌──────────────────┐
                 │  最终评审报告     │
                 │  含：建议、风险、 │
                 │  实施路线图       │
                 └──────────────────┘