🌐 Multi-Agent 系统概述

"""
Subagents 模式 - 专业化分工
"""
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, MessagesState, START

# 定义专门化的子代理节点
def research_agent(state: MessagesState):
    """研究专员：负责信息收集和分析"""
    # 使用搜索工具收集信息
    return {"messages": ["研究结果：已收集相关数据..."]}

def writer_agent(state: MessagesState):
    """写作专员：负责内容创作"""
    # 基于研究结果创作内容
    return {"messages": ["文章草稿：基于研究数据撰写..."]}

def editor_agent(state: MessagesState):
    """编辑专员：负责内容审核和优化"""
    # 审核和优化内容
    return {"messages": ["最终稿件：已优化并审核完成"]}

# 构建工作流
graph = StateGraph(MessagesState)
graph.add_node("researcher", research_agent)
graph.add_node("writer", writer_agent)
graph.add_node("editor", editor_agent)

# 定义执行顺序
graph.add_edge(START, "researcher")
graph.add_edge("researcher", "writer")
graph.add_edge("writer", "editor")

app = graph.compile()

# 使用
result = app.invoke({
    "messages": [HumanMessage(content="写一篇关于 AI 的文章")]
})

# 优势：
# - 每个代理专注于自己的领域
# - 易于维护和扩展
# - 可以为不同代理配置不同的模型和工具

"""
Handoffs 模式 - 动态交接控制
"""
from typing import Literal
from langgraph.graph import StateGraph, MessagesState

def customer_service_agent(state: MessagesState):
    """客服代理：处理常规问题"""
    messages = state["messages"]
    last_message = messages[-1].content

    # 判断是否需要转给技术支持
    if "技术问题" in last_message or "bug" in last_message:
        return {
            "messages": ["我将为您转接技术支持团队"],
            "next": "technical_support"  # 交接指令
        }

    return {
        "messages": ["问题已解决"],
        "next": "end"
    }

def technical_support_agent(state: MessagesState):
    """技术支持代理：处理技术问题"""
    messages = state["messages"]
    last_message = messages[-1].content

    # 判断是否需要升级到工程师
    if "严重" in last_message or "紧急" in last_message:
        return {
            "messages": ["问题已升级，转接工程师"],
            "next": "engineer"
        }

    return {
        "messages": ["技术问题已解决"],
        "next": "end"
    }

def engineer_agent(state: MessagesState):
    """工程师代理：处理复杂技术问题"""
    return {
        "messages": ["工程师已介入，正在处理"],
        "next": "end"
    }

# 构建带交接的工作流
def route_next(state: MessagesState) -> Literal["customer_service", "technical_support", "engineer", "end"]:
    """根据 next 字段路由"""
    return state.get("next", "customer_service")

graph = StateGraph(MessagesState)
graph.add_node("customer_service", customer_service_agent)
graph.add_node("technical_support", technical_support_agent)
graph.add_node("engineer", engineer_agent)

# 添加条件路由
graph.add_conditional_edges("customer_service", route_next)
graph.add_conditional_edges("technical_support", route_next)

app = graph.compile()

# 优势：
# - 动态决策何时交接
# - 支持多层级升级
# - 保持上下文连续性

"""
Skills 模式 - 共享技能系统
"""
from langchain.tools import tool

# 定义共享技能
@tool
def search_database(query: str) -> dict:
    """搜索数据库的共享技能"""
    # 所有代理都可以使用此技能
    return {"results": ["数据 1", "数据 2", "数据 3"]}

@tool
def send_notification(recipient: str, message: str) -> str:
    """发送通知的共享技能"""
    # 所有代理都可以发送通知
    return f"通知已发送给 {recipient}"

@tool
def format_report(data: dict, format: str) -> str:
    """格式化报告的共享技能"""
    # 所有代理都可以使用统一的格式化
    return f"报告已格式化为 {format}"

# 为多个代理配置相同的技能
common_skills = [search_database, send_notification, format_report]

# Agent 1: 数据分析师（使用共享技能）
from langchain_core.agents import create_agent

analyst_agent = create_agent(
    model="gpt-4o",
    tools=common_skills,  # 共享技能
    system_prompt="你是数据分析师，可以搜索数据、发送通知、格式化报告"
)

# Agent 2: 报告生成器（使用相同的共享技能）
reporter_agent = create_agent(
    model="claude-sonnet-4-5",
    tools=common_skills,  # 相同的共享技能
    system_prompt="你是报告生成器，可以搜索数据、发送通知、格式化报告"
)

# 优势：
# - 避免重复定义工具
# - 保证工具行为一致性
# - 易于统一维护和升级

"""
Router 模式 - 智能路由分发
"""
from typing import Literal
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState

# 定义路由决策系统
route_prompt = ChatPromptTemplate.from_messages([
    ("system", """你是路由助手。根据用户请求，决定分配给哪个专员：

- sales: 销售咨询、产品介绍、报价
- support: 售后问题、技术支持、故障排除
- billing: 账单查询、支付问题、发票
- general: 其他一般性问题

仅返回一个词：sales, support, billing, 或 general"""),
    ("human", "{input}")
])

router_llm = ChatOpenAI(model="gpt-4o")
router_chain = route_prompt | router_llm

def router(state: MessagesState) -> Literal["sales", "support", "billing", "general"]:
    """智能路由器"""
    user_input = state["messages"][-1].content
    result = router_chain.invoke({"input": user_input})
    return result.content.strip().lower()

# 定义各个专员代理
def sales_agent(state: MessagesState):
    """销售专员"""
    return {"messages": ["销售专员：我来帮您介绍产品和报价"]}

def support_agent(state: MessagesState):
    """技术支持专员"""
    return {"messages": ["技术支持：我来帮您解决技术问题"]}

def billing_agent(state: MessagesState):
    """账单专员"""
    return {"messages": ["账单专员：我来帮您处理账单问题"]}

def general_agent(state: MessagesState):
    """通用专员"""
    return {"messages": ["客服：我来为您服务"]}

# 构建路由工作流
graph = StateGraph(MessagesState)

# 添加所有代理节点
graph.add_node("sales", sales_agent)
graph.add_node("support", support_agent)
graph.add_node("billing", billing_agent)
graph.add_node("general", general_agent)

# 添加条件路由（从起点根据内容路由）
graph.add_conditional_edges(START, router)

app = graph.compile()

# 测试路由
test_cases = [
    "我想了解你们的产品价格",      # → sales
    "我的账单有问题",              # → billing
    "软件崩溃了",                  # → support
    "你们的营业时间是什么？"        # → general
]

for test in test_cases:
    result = app.invoke({"messages": [HumanMessage(content=test)]})
    print(f"请求: {test}")
    print(f"响应: {result['messages'][-1]}\n")

# 优势：
# - 自动分配到最合适的代理
# - 减少人工分类工作
# - 提高响应效率

"""
Custom Workflow 模式 - 复杂工作流编排
"""
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# 定义工作流状态
class WorkflowState(TypedDict):
    messages: Annotated[list, add_messages]
    research_data: str
    analysis_result: str
    report_draft: str
    approved: bool

# 步骤 1: 研究代理
def research_phase(state: WorkflowState):
    """研究阶段：收集数据"""
    return {
        "research_data": "已收集的研究数据...",
        "messages": ["研究阶段完成"]
    }

# 步骤 2: 分析代理
def analysis_phase(state: WorkflowState):
    """分析阶段：数据分析"""
    return {
        "analysis_result": "分析结果：关键发现...",
        "messages": ["分析阶段完成"]
    }

# 步骤 3: 写作代理
def writing_phase(state: WorkflowState):
    """写作阶段：生成报告"""
    return {
        "report_draft": "报告草稿：基于分析结果...",
        "messages": ["报告草稿完成"]
    }

# 步骤 4: 人工审批（模拟）
def human_review(state: WorkflowState):
    """人工审批阶段"""
    # 实际应用中会暂停等待人工审批
    # 这里模拟自动批准
    return {
        "approved": True,
        "messages": ["报告已批准"]
    }

# 步骤 5: 发布代理
def publish_phase(state: WorkflowState):
    """发布阶段：格式化和发布"""
    return {
        "messages": ["报告已发布"]
    }

# 决策节点：根据审批结果决定下一步
def should_publish(state: WorkflowState) -> Literal["publish", "revise"]:
    """决策：是否发布"""
    return "publish" if state.get("approved") else "revise"

# 步骤 6: 修订代理
def revise_phase(state: WorkflowState):
    """修订阶段：根据反馈修改"""
    return {
        "report_draft": "修订后的报告...",
        "messages": ["报告已修订，重新提交审批"]
    }

# 构建复杂工作流
workflow = StateGraph(WorkflowState)

# 添加所有节点
workflow.add_node("research", research_phase)
workflow.add_node("analysis", analysis_phase)
workflow.add_node("writing", writing_phase)
workflow.add_node("review", human_review)
workflow.add_node("publish", publish_phase)
workflow.add_node("revise", revise_phase)

# 定义流程路径
workflow.add_edge(START, "research")
workflow.add_edge("research", "analysis")
workflow.add_edge("analysis", "writing")
workflow.add_edge("writing", "review")

# 条件分支：根据审批结果
workflow.add_conditional_edges(
    "review",
    should_publish,
    {
        "publish": "publish",
        "revise": "revise"
    }
)

# 修订后重新审批
workflow.add_edge("revise", "review")

# 发布后结束
workflow.add_edge("publish", END)

# 编译工作流
app = workflow.compile()

# 使用
result = app.invoke({
    "messages": [HumanMessage(content="创建季度销售报告")]
})

print("工作流执行完成")
print("最终消息:", result["messages"])

# 工作流执行路径：
# 研究 → 分析 → 写作 → 审批 → 发布（如批准）
#                     ↓
#                   修订 → 重新审批（如未批准）

# 优势：
# - 灵活定义复杂业务流程
# - 支持条件分支和循环
# - 易于可视化和维护
# - 可集成人工审批节点

# 错误处理模式
def robust_agent(state: MessagesState):
    """带错误处理的代理"""
    try:
        # 尝试主要功能
        result = primary_function(state)
        return result
    except SpecificError as e:
        # 特定错误的降级策略
        logger.error(f"主功能失败: {e}")
        return fallback_function(state)
    except Exception as e:
        # 通用错误处理
        logger.error(f"未预期错误: {e}")
        return {
            "messages": ["抱歉，遇到问题。请稍后重试。"],
            "error": str(e)
        }

# 优雅降级示例
def search_with_fallback(query: str):
    """带降级的搜索"""
    try:
        # 尝试主搜索引擎
        return primary_search(query)
    except:
        # 降级到备用搜索
        return fallback_search(query)

# 使用 LangSmith 追踪
import os

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "multi-agent-system"

# 所有代理交互会被自动记录到 LangSmith
# 可以查看：
# - 代理调用序列
# - 每个代理的输入输出
# - 执行时间和成本
# - 错误和异常