10e - 多 Agent 实战案例

                    ┌──────────────────────────────────────┐
                    │          ContentState（共享状态）       │
                    │  topic, research, draft, review,     │
                    │  final_content, revision_count       │
                    └──────────────────────────────────────┘
                                     │
                    ┌────────────────┼────────────────┐
                    ▼                ▼                ▼
              ┌──────────┐   ┌──────────┐   ┌──────────┐
              │ 调研节点  │──▶│ 撰写节点  │──▶│ 审校节点  │
              │Researcher│   │ Writer   │   │ Reviewer │
              └──────────┘   └──────────┘   └──────────┘
                                  ▲                │
                                  │    ┌───────────┤
                                  │    ▼           ▼
                              [不通过]        [通过]
                              回退重写     ┌──────────┐
                                          │ 定稿节点  │
                                          │Publisher │
                                          └──────────┘
                                               │
                                               ▼
                                             END

"""
LangGraph 多 Agent 内容生产管线
要求：pip install langgraph langchain-openai
"""
 
import operator
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
 
# ============================================================
# 1. 定义共享状态
# ============================================================
class ContentState(TypedDict):
    """内容生产管线的共享状态"""
    topic: str                                    # 输入主题
    research: str                                 # 调研结果
    draft: str                                    # 草稿内容
    review_feedback: str                          # 审校反馈
    review_passed: bool                           # 审校是否通过
    final_content: str                            # 最终定稿
    revision_count: Annotated[int, operator.add]  # 修订次数（自动累加）
 
# ============================================================
# 2. 初始化 LLM
# ============================================================
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
 
# ============================================================
# 3. 定义各 Agent 节点函数
# ============================================================
def researcher_node(state: ContentState) -> dict:
    """调研节点：根据主题收集背景信息和关键要点"""
    topic = state["topic"]
    prompt = f"""你是一位资深调研分析师。请针对以下主题进行深度调研，
输出结构化的调研报告，包含：
1. 主题背景概述
2. 核心要点（至少 5 个）
3. 最新趋势和数据
4. 潜在的争议或不同观点
 
主题：{topic}
"""
    response = llm.invoke(prompt)
    return {"research": response.content}
 
 
def writer_node(state: ContentState) -> dict:
    """撰写节点：基于调研结果撰写文章草稿"""
    topic = state["topic"]
    research = state["research"]
    feedback = state.get("review_feedback", "")
 
    revision_hint = ""
    if feedback:
        revision_hint = f"\n\n上一版审校反馈（请据此修改）：\n{feedback}"
 
    prompt = f"""你是一位专业内容撰稿人。请基于以下调研资料撰写一篇高质量文章。
 
要求：
- 标题吸引人
- 结构清晰（引言、正文分段、结论）
- 语言流畅，适合技术博客读者
- 字数 800-1200 字
{revision_hint}
 
主题：{topic}
调研资料：{research}
"""
    response = llm.invoke(prompt)
    return {"draft": response.content, "revision_count": 1}
 
 
def reviewer_node(state: ContentState) -> dict:
    """审校节点：审查草稿质量，决定通过或打回"""
    draft = state["draft"]
    revision_count = state.get("revision_count", 0)
 
    prompt = f"""你是一位严格的内容审校编辑。请审查以下文章草稿，
从以下维度评分（1-10）：
1. 内容准确性
2. 结构逻辑性
3. 语言表达质量
4. 读者价值
 
如果所有维度均 ≥ 7 分，输出 "APPROVED"。
否则输出 "REVISION_NEEDED" 并给出具体修改建议。
 
当前修订轮次：{revision_count}（最多允许 3 轮）
 
文章草稿：
{draft}
"""
    response = llm.invoke(prompt)
    content = response.content
 
    passed = "APPROVED" in content or revision_count >= 3
    return {
        "review_feedback": content,
        "review_passed": passed,
    }
 
 
def publisher_node(state: ContentState) -> dict:
    """定稿节点：最终润色并输出成品"""
    draft = state["draft"]
    prompt = f"""你是一位出版编辑。请对以下已审核通过的文章做最终润色：
- 检查错别字和语法
- 优化段落过渡
- 添加适当的 emoji 和格式标记
- 确保开头和结尾有力
 
文章：
{draft}
"""
    response = llm.invoke(prompt)
    return {"final_content": response.content}
 
# ============================================================
# 4. 定义条件路由函数
# ============================================================
def should_revise(state: ContentState) -> str:
    """根据审校结果决定下一步：通过则定稿，否则回退重写"""
    if state.get("review_passed", False):
        return "publish"
    return "revise"
 
# ============================================================
# 5. 构建状态图
# ============================================================
workflow = StateGraph(ContentState)
 
# 添加节点
workflow.add_node("researcher", researcher_node)
workflow.add_node("writer", writer_node)
workflow.add_node("reviewer", reviewer_node)
workflow.add_node("publisher", publisher_node)
 
# 添加边：定义流程
workflow.set_entry_point("researcher")
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "reviewer")
 
# 条件边：审校通过 → 定稿，不通过 → 回退重写
workflow.add_conditional_edges(
    "reviewer",
    should_revise,
    {
        "publish": "publisher",
        "revise": "writer",       # 回退到撰写节点
    },
)
workflow.add_edge("publisher", END)
 
# 编译图
app = workflow.compile()
 
# ============================================================
# 6. 执行管线
# ============================================================
if __name__ == "__main__":
    result = app.invoke({
        "topic": "2025 年 AI Agent 框架生态全景分析",
        "research": "",
        "draft": "",
        "review_feedback": "",
        "review_passed": False,
        "final_content": "",
        "revision_count": 0,
    })
 
    print("=" * 60)
    print("📝 最终文章：")
    print("=" * 60)
    print(result["final_content"])
    print(f"\n📊 总修订轮次：{result['revision_count']}")

[Researcher] → 输出 2000 字调研报告，包含 6 个核心要点
[Writer]     → 基于调研撰写 1000 字初稿
[Reviewer]   → 评分：准确性 8、逻辑性 6、表达 7、价值 7 → REVISION_NEEDED
[Writer]     → 根据反馈修改，加强逻辑结构（第 2 轮）
[Reviewer]   → 评分：准确性 8、逻辑性 8、表达 8、价值 8 → APPROVED
[Publisher]  → 最终润色输出

┌─────────────────────────────────────────────────────┐
│                    Crew（团队）                       │
│                 process=sequential                   │
│                                                     │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐ │
│  │ SEO 策略师   │  │ 技术作者     │  │ 编辑        │ │
│  │             │  │             │  │             │ │
│  │ role:       │  │ role:       │  │ role:       │ │
│  │ SEO策略师   │  │ 高级技术作者 │  │ 内容编辑    │ │
│  │             │  │             │  │             │ │
│  │ goal:       │  │ goal:       │  │ goal:       │ │
│  │ 找到最佳    │  │ 撰写高质量  │  │ 确保文章    │ │
│  │ 关键词策略  │  │ 技术文章    │  │ 完美无瑕    │ │
│  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘ │
│         │                │                │        │
│         ▼                ▼                ▼        │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐ │
│  │ Task 1:     │  │ Task 2:     │  │ Task 3:     │ │
│  │ 关键词研究  │──▶│ 文章撰写    │──▶│ 编辑润色    │ │
│  │ & SEO 策略  │  │             │  │ & SEO 优化  │ │
│  └─────────────┘  └─────────────┘  └─────────────┘ │
│                                                     │
└─────────────────────────────────────────────────────┘

"""
CrewAI 技术博客自动化团队
要求：pip install crewai crewai-tools
"""
 
from crewai import Agent, Task, Crew, Process
 
# ============================================================
# 1. 定义 Agent（角色）
# ============================================================
seo_strategist = Agent(
    role="SEO 策略师",
    goal="为给定主题找到最佳关键词策略和内容角度，最大化搜索引擎可见性",
    backstory="""你是一位拥有 8 年经验的 SEO 专家，精通技术内容的搜索优化。
你深谙 Google 算法更新趋势，擅长发现长尾关键词机会。
你的策略曾帮助多个技术博客实现月均 300% 的有机流量增长。""",
    verbose=True,
    allow_delegation=False,
    llm="gpt-4o",
)
 
tech_writer = Agent(
    role="高级技术作者",
    goal="基于 SEO 策略和调研资料，撰写深度、准确、易读的技术博客文章",
    backstory="""你是一位全栈开发者转型的技术作者，拥有 5 年技术写作经验。
你擅长将复杂的技术概念用通俗易懂的语言解释清楚，
同时保持技术深度。你的文章风格兼具专业性和可读性。""",
    verbose=True,
    allow_delegation=False,
    llm="gpt-4o",
)
 
editor = Agent(
    role="内容编辑",
    goal="确保文章在语言质量、技术准确性和 SEO 优化三个维度都达到发布标准",
    backstory="""你是一位资深内容编辑，曾在多家技术媒体担任主编。
你对文字有极高的要求，同时理解 SEO 的重要性。
你的审校标准：零错别字、逻辑通顺、SEO 友好、读者价值最大化。""",
    verbose=True,
    allow_delegation=False,
    llm="gpt-4o",
)
 
# ============================================================
# 2. 定义 Task（任务）
# ============================================================
keyword_research_task = Task(
    description="""针对主题「{topic}」进行全面的 SEO 关键词研究：
 
1. 识别 1 个主关键词和 5-8 个长尾关键词
2. 分析每个关键词的搜索意图（信息型/导航型/交易型）
3. 评估竞争难度（低/中/高）
4. 建议文章标题（3 个备选）
5. 建议文章大纲结构
6. 推荐内链和外链策略
 
输出格式：结构化的 SEO 策略报告""",
    expected_output="包含关键词列表、搜索意图分析、标题建议和大纲的 SEO 策略报告",
    agent=seo_strategist,
)
 
writing_task = Task(
    description="""基于 SEO 策略师提供的关键词策略和大纲，撰写一篇完整的技术博客文章：
 
要求：
1. 字数 1500-2500 字
2. 自然融入主关键词（密度 1-2%）和长尾关键词
3. 包含引言、至少 3 个主体段落、结论
4. 每个段落有清晰的小标题（H2/H3）
5. 包含至少 1 个代码示例或技术图示
6. 包含至少 2 个实际应用场景
7. 语言风格：专业但不晦涩，适合中高级开发者
 
主题：「{topic}」""",
    expected_output="一篇完整的、SEO 优化的技术博客文章（Markdown 格式）",
    agent=tech_writer,
    context=[keyword_research_task],  # 依赖 SEO 策略任务的输出
)
 
editing_task = Task(
    description="""对技术作者提交的文章进行全面编辑和优化：
 
审校维度：
1. 语言质量：修正错别字、语法错误、表达不当
2. 技术准确性：验证技术描述和代码示例的正确性
3. SEO 优化：检查关键词分布、meta description、标题标签
4. 结构优化：确保逻辑流畅、段落过渡自然
5. 读者体验：添加要点总结、行动建议
 
输出：最终可发布版本的文章 + 编辑修改说明""",
    expected_output="最终润色完成的文章（Markdown 格式）和编辑修改说明",
    agent=editor,
    context=[keyword_research_task, writing_task],  # 依赖前两个任务
)
 
# ============================================================
# 3. 组建 Crew 并执行
# ============================================================
blog_crew = Crew(
    agents=[seo_strategist, tech_writer, editor],
    tasks=[keyword_research_task, writing_task, editing_task],
    process=Process.sequential,  # 顺序执行
    verbose=True,
)
 
if __name__ == "__main__":
    result = blog_crew.kickoff(
        inputs={"topic": "如何用 LangGraph 构建生产级多 Agent 系统"}
    )
 
    print("=" * 60)
    print("📝 最终文章：")
    print("=" * 60)
    print(result.raw)
    print(f"\n📊 Token 使用统计：{result.token_usage}")

[SEO 策略师] 开始执行：关键词研究 & SEO 策略
  → 主关键词："LangGraph 多 Agent"
  → 长尾关键词："LangGraph 生产部署"、"多 Agent 架构设计"...
  → 推荐标题："从零到生产：用 LangGraph 构建多 Agent 系统完全指南"
  → 输出 SEO 策略报告（约 800 字）

[高级技术作者] 开始执行：文章撰写
  → 接收 SEO 策略报告作为上下文
  → 输出 2000 字技术文章（含代码示例）

[内容编辑] 开始执行：编辑润色 & SEO 优化
  → 接收 SEO 策略 + 文章草稿作为上下文
  → 修正 3 处表达问题，优化 2 处 SEO 关键词分布
  → 输出最终版本 + 修改说明

                         用户提问
                            │
                            ▼
                    ┌───────────────┐
                    │  分诊 Agent    │
                    │  (Triage)     │
                    │               │
                    │ 判断问题类型   │
                    └───────┬───────┘
                            │
              ┌─────────────┼─────────────┐
              ▼             ▼             ▼
     ┌──────────────┐ ┌──────────┐ ┌──────────────┐
     │ 技术支持      │ │ 账单咨询 │ │ 销售咨询      │
     │ Agent        │ │ Agent    │ │ Agent        │
     │              │ │          │ │              │
     │ 工具：       │ │ 工具：   │ │ 工具：       │
     │ - 查知识库   │ │ - 查账单 │ │ - 查产品目录 │
     │ - 查工单     │ │ - 退款   │ │ - 生成报价   │
     └──────────────┘ └──────────┘ └──────────────┘
              │             │             │
              ▼             ▼             ▼
                      返回结果给用户

"""
OpenAI Agents SDK 智能客服路由系统
要求：pip install openai-agents
"""
 
from agents import Agent, Runner, function_tool, handoff
 
# ============================================================
# 1. 定义工具函数
# ============================================================
@function_tool
def search_knowledge_base(query: str) -> str:
    """搜索技术知识库，查找问题的解决方案"""
    # 实际项目中连接向量数据库或搜索引擎
    knowledge = {
        "登录失败": "请尝试：1) 清除浏览器缓存 2) 重置密码 3) 检查账号是否被锁定",
        "API 超时": "常见原因：1) 网络问题 2) 请求量超限 3) 服务端维护。建议检查状态页面。",
        "数据导出": "进入设置 → 数据管理 → 导出，支持 CSV 和 JSON 格式。",
    }
    for key, value in knowledge.items():
        if key in query:
            return value
    return "未找到直接匹配的解决方案，建议提交工单由人工处理。"
 
 
@function_tool
def lookup_ticket(ticket_id: str) -> str:
    """查询工单状态"""
    return f"工单 {ticket_id} 状态：处理中，预计 24 小时内回复。"
 
 
@function_tool
def check_billing(customer_id: str) -> str:
    """查询客户账单信息"""
    return f"客户 {customer_id} 当前账单：Pro 计划 $49/月，下次扣费日期 2025-08-01，无欠费。"
 
 
@function_tool
def process_refund(customer_id: str, amount: float, reason: str) -> str:
    """处理退款申请"""
    return f"退款申请已提交：客户 {customer_id}，金额 ${amount}，原因：{reason}。预计 3-5 个工作日到账。"
 
 
@function_tool
def get_product_catalog(category: str) -> str:
    """获取产品目录"""
    catalog = {
        "基础版": "Free 计划：5 用户，1GB 存储，社区支持",
        "专业版": "Pro 计划：$49/月，50 用户，100GB 存储，优先支持",
        "企业版": "Enterprise 计划：联系销售，无限用户，无限存储，专属客户经理",
    }
    return "\n".join(f"- {k}: {v}" for k, v in catalog.items())
 
 
@function_tool
def generate_quote(plan: str, seats: int, duration_months: int) -> str:
    """生成报价单"""
    prices = {"Pro": 49, "Enterprise": 199}
    base = prices.get(plan, 49)
    total = base * seats * duration_months
    discount = 0.1 if duration_months >= 12 else 0
    final = total * (1 - discount)
    return f"报价单：{plan} 计划，{seats} 席位，{duration_months} 个月 → 总价 ${final:.0f}（折扣 {discount*100:.0f}%）"
 
# ============================================================
# 2. 定义专业 Agent
# ============================================================
tech_support_agent = Agent(
    name="技术支持专家",
    instructions="""你是一位耐心、专业的技术支持工程师。
你的职责是帮助用户解决技术问题。
 
工作流程：
1. 先搜索知识库查找已知解决方案
2. 如果找到方案，清晰地指导用户操作
3. 如果未找到，帮助用户提交工单
4. 始终保持友好和专业的语气
 
注意：你只处理技术问题，如果用户问账单或销售问题，告知他们需要转接。""",
    tools=[search_knowledge_base, lookup_ticket],
)
 
billing_agent = Agent(
    name="账单咨询专家",
    instructions="""你是一位细心的账单咨询专家。
你的职责是帮助用户处理账单相关问题。
 
你可以：
1. 查询账单详情和付款历史
2. 处理退款申请（需要客户确认）
3. 解释计费规则和套餐差异
 
注意：退款金额超过 $500 需要提示用户联系人工客服。
你只处理账单问题，技术问题和销售问题请告知需要转接。""",
    tools=[check_billing, process_refund],
)
 
sales_agent = Agent(
    name="销售咨询专家",
    instructions="""你是一位热情但不过度推销的销售顾问。
你的职责是帮助潜在客户了解产品并生成报价。
 
工作流程：
1. 了解客户需求（团队规模、使用场景、预算）
2. 推荐合适的产品方案
3. 生成定制报价
4. 解答产品功能问题
 
注意：年付享 10% 折扣。企业版需要联系销售经理做最终确认。""",
    tools=[get_product_catalog, generate_quote],
)
 
# ============================================================
# 3. 定义分诊 Agent（路由器）
# ============================================================
triage_agent = Agent(
    name="智能客服分诊",
    instructions="""你是智能客服系统的分诊 Agent。你的唯一职责是判断用户问题的类型，
并将用户交接给正确的专业 Agent。
 
分类规则：
- 技术问题（登录、API、bug、功能使用、数据导出等）→ 交接给「技术支持专家」
- 账单问题（付款、退款、发票、套餐变更、扣费等）→ 交接给「账单咨询专家」
- 销售问题（产品咨询、报价、试用、升级、功能对比等）→ 交接给「销售咨询专家」
 
注意：
- 不要自己回答问题，你的职责只是路由
- 如果无法判断类型，礼貌地询问用户更多信息
- 交接时简要告知用户正在转接""",
    handoffs=[
        handoff(agent=tech_support_agent, description="处理技术支持相关问题"),
        handoff(agent=billing_agent, description="处理账单和付款相关问题"),
        handoff(agent=sales_agent, description="处理销售咨询和报价相关问题"),
    ],
)
 
# ============================================================
# 4. 运行客服系统
# ============================================================
async def handle_customer_query(query: str):
    """处理单个客户查询"""
    print(f"\n{'='*50}")
    print(f"👤 客户提问：{query}")
    print(f"{'='*50}")
 
    result = await Runner.run(
        starting_agent=triage_agent,
        input=query,
    )
 
    print(f"\n🤖 最终回复（by {result.last_agent.name}）：")
    print(result.final_output)
    return result
 
 
if __name__ == "__main__":
    import asyncio
 
    queries = [
        "我的 API 调用一直超时，怎么解决？",
        "我想了解一下你们的企业版方案，我们团队有 50 人",
        "上个月的账单好像多扣了钱，能帮我查一下吗？",
    ]
 
    async def main():
        for query in queries:
            await handle_customer_query(query)
            print("\n" + "-" * 60)
 
    asyncio.run(main())

查询 1："我的 API 调用一直超时，怎么解决？"
  [分诊 Agent] → 识别为技术问题 → handoff 给「技术支持专家」
  [技术支持专家] → 调用 search_knowledge_base("API 超时")
  → 回复：常见原因和排查步骤

查询 2："我想了解企业版方案，团队 50 人"
  [分诊 Agent] → 识别为销售问题 → handoff 给「销售咨询专家」
  [销售咨询专家] → 调用 get_product_catalog("企业版")
                 → 调用 generate_quote("Enterprise", 50, 12)
  → 回复：产品介绍 + 定制报价

查询 3："上个月账单多扣了钱"
  [分诊 Agent] → 识别为账单问题 → handoff 给「账单咨询专家」
  [账单咨询专家] → 调用 check_billing(customer_id)
  → 回复：账单详情 + 后续处理建议

你的多 Agent 系统属于哪种模式？

A. 流水线模式（阶段明确、有条件分支和回退）
   → 选择 LangGraph

B. 团队协作模式（角色分工、任务依赖、类似真实团队）
   → 选择 CrewAI

C. 路由分发模式（一个入口、多个专业处理器）
   → 选择 OpenAI Agents SDK

# LangGraph 项目初始化
pip install langgraph langchain-openai
# 核心概念：StateGraph → add_node → add_edge → compile → invoke
 
# CrewAI 项目初始化
pip install crewai
# 核心概念：Agent(role, goal, backstory) → Task(description) → Crew.kickoff()
 
# OpenAI Agents SDK 项目初始化
pip install openai-agents
# 核心概念：Agent(instructions, tools, handoffs) → Runner.run()

你是一位 AI 架构师，请帮我设计一个多 Agent 系统。

场景描述：[描述你的业务场景]
Agent 数量：[预估需要几个 Agent]
协作模式：[流水线 / 团队协作 / 路由分发 / 混合]
技术约束：[使用的 LLM / 预算限制 / 延迟要求]

请输出：
1. 系统架构图（ASCII 或 Mermaid）
2. 每个 Agent 的角色定义和工具列表
3. Agent 间的通信流程
4. 推荐使用的框架及理由
5. 预估的 token 消耗和成本