🔄 Multi-Agent 交接机制

"""
基于状态的交接 - 简单两个代理
"""
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage

# 定义状态（包含交接目标）
class HandoffState(TypedDict):
    messages: Annotated[list, add_messages]
    next: str  # 指定下一个代理

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

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

    return {
        "messages": [HumanMessage(content="问题已解决")],
        "next": "end"
    }

# 代理 2：技术支持
def technical_support(state: HandoffState):
    """技术支持代理"""
    return {
        "messages": [HumanMessage(content="技术支持：我来帮您解决技术问题")],
        "next": "end"
    }

# 路由函数
def route_next(state: HandoffState) -> Literal["customer_service", "technical_support", "end"]:
    """根据 next 字段路由"""
    next_agent = state.get("next", "customer_service")
    if next_agent == "end":
        return END
    return next_agent

# 构建工作流
graph = StateGraph(HandoffState)
graph.add_node("customer_service", customer_service)
graph.add_node("technical_support", technical_support)

# 初始路由
graph.add_edge(START, "customer_service")

# 条件路由
graph.add_conditional_edges(
    "customer_service",
    route_next,
    {
        "technical_support": "technical_support",
        "end": END
    }
)
graph.add_conditional_edges(
    "technical_support",
    route_next,
    {"end": END}
)

app = graph.compile()

# 测试
result = app.invoke({
    "messages": [HumanMessage(content="我遇到了技术问题")]
})

# 输出：
# 1. 客服：我将为您转接技术支持团队
# 2. 技术支持：我来帮您解决技术问题

# 优势：
# - 简单直观
# - 代理自主决定交接
# - 易于理解和调试

"""
多层级升级交接 - 客服 → 技术 → 工程师
"""
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage

class EscalationState(TypedDict):
    messages: Annotated[list, add_messages]
    severity: str  # "low", "medium", "high"
    next: str

# L1: 客服代理
def l1_support(state: EscalationState):
    """L1 客服：处理常规问题"""
    messages = state["messages"]
    last_msg = messages[-1].content

    # 评估严重性
    if "紧急" in last_msg or "严重" in last_msg:
        severity = "high"
        next_agent = "l2_technical"
        response = "这是紧急问题，我将为您转接技术支持（L2）"
    elif "技术" in last_msg or "bug" in last_msg.lower():
        severity = "medium"
        next_agent = "l2_technical"
        response = "这需要技术支持，我将为您转接（L2）"
    else:
        severity = "low"
        next_agent = "end"
        response = "问题已解决"

    return {
        "messages": [AIMessage(content=f"L1 客服：{response}")],
        "severity": severity,
        "next": next_agent
    }

# L2: 技术支持代理
def l2_technical(state: EscalationState):
    """L2 技术支持：诊断技术问题"""
    severity = state.get("severity", "medium")

    # 高严重性问题升级给工程师
    if severity == "high":
        return {
            "messages": [AIMessage(content="L2 技术支持：这是高优先级问题，升级给工程师（L3）")],
            "next": "l3_engineer"
        }

    return {
        "messages": [AIMessage(content="L2 技术支持：问题已诊断和解决")],
        "next": "end"
    }

# L3: 工程师代理
def l3_engineer(state: EscalationState):
    """L3 工程师：深度排查和修复"""
    return {
        "messages": [AIMessage(content="L3 工程师：已深度排查并修复问题")],
        "next": "end"
    }

# 路由函数
def route(state: EscalationState) -> Literal["l1_support", "l2_technical", "l3_engineer", "end"]:
    next_agent = state.get("next", "l1_support")
    return END if next_agent == "end" else next_agent

# 构建三层升级工作流
graph = StateGraph(EscalationState)
graph.add_node("l1_support", l1_support)
graph.add_node("l2_technical", l2_technical)
graph.add_node("l3_engineer", l3_engineer)

graph.add_edge(START, "l1_support")
graph.add_conditional_edges("l1_support", route)
graph.add_conditional_edges("l2_technical", route)
graph.add_conditional_edges("l3_engineer", route)

app = graph.compile()

# 测试不同严重性
test_cases = [
    "我有一个简单的问题",           # L1 处理
    "遇到了技术bug",                # L1 → L2
    "紧急！系统崩溃了！",           # L1 → L2 → L3
]

for test in test_cases:
    print(f"\n测试: {test}")
    result = app.invoke({"messages": [HumanMessage(content=test)]})
    for msg in result["messages"]:
        if isinstance(msg, AIMessage):
            print(f"  {msg.content}")

# 输出示例：
# 测试: 紧急！系统崩溃了！
#   L1 客服：这是紧急问题，我将为您转接技术支持（L2）
#   L2 技术支持：这是高优先级问题，升级给工程师（L3）
#   L3 工程师：已深度排查并修复问题

# 优势：
# - 自动升级机制
# - 基于严重性分级
# - 每层专注于自己的职责

"""
带状态传递的交接 - 传递上下文和数据
"""
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class ContextualHandoffState(TypedDict):
    messages: Annotated[list, add_messages]
    user_id: str
    issue_type: str
    collected_data: dict  # 收集的诊断数据
    next: str

# 代理 1：诊断代理
def diagnostic_agent(state: ContextualHandoffState):
    """诊断代理：收集问题信息"""
    # 收集诊断数据
    collected_data = {
        "symptoms": "应用崩溃",
        "frequency": "频繁",
        "last_occurrence": "2 小时前",
        "error_logs": ["Error: NullPointerException", "Error: OutOfMemory"]
    }

    # 判断问题类型
    issue_type = "critical_bug"

    return {
        "messages": [HumanMessage(content="已收集诊断信息，转接给bug修复团队")],
        "issue_type": issue_type,
        "collected_data": collected_data,
        "next": "bug_fix"
    }

# 代理 2：Bug 修复代理
def bug_fix_agent(state: ContextualHandoffState):
    """Bug修复代理：基于诊断数据修复"""
    # 读取传递的数据
    data = state["collected_data"]
    issue_type = state["issue_type"]

    # 基于诊断数据执行修复
    fix_plan = f"""
    问题类型：{issue_type}
    症状：{data['symptoms']}
    错误日志：{', '.join(data['error_logs'])}

    修复方案：
    1. 分析 NullPointerException 根因
    2. 检查内存泄漏
    3. 部署修复补丁
    """

    return {
        "messages": [HumanMessage(content=f"Bug修复团队：{fix_plan}")],
        "next": "end"
    }

# 路由
def route(state: ContextualHandoffState) -> Literal["diagnostic", "bug_fix", "end"]:
    next_agent = state.get("next", "diagnostic")
    return END if next_agent == "end" else next_agent

# 构建工作流
graph = StateGraph(ContextualHandoffState)
graph.add_node("diagnostic", diagnostic_agent)
graph.add_node("bug_fix", bug_fix_agent)

graph.add_edge(START, "diagnostic")
graph.add_conditional_edges("diagnostic", route)
graph.add_conditional_edges("bug_fix", route)

app = graph.compile()

# 使用
result = app.invoke({
    "messages": [HumanMessage(content="应用一直崩溃")],
    "user_id": "user123"
})

# 优势：
# - 完整的上下文传递
# - 后续代理可以利用之前收集的数据
# - 避免重复询问用户
# - 专业化分工（诊断 vs 修复）

"""
完整实战示例 - 智能客服系统（多层级交接）
"""
import uuid
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

# ========== 1. 定义状态 ==========

class SupportState(TypedDict):
    messages: Annotated[list, add_messages]
    user_id: str
    session_id: str

    # 问题信息
    issue_type: str  # "general", "technical", "critical"
    priority: str    # "low", "medium", "high"

    # 诊断数据
    symptoms: list[str]
    error_logs: list[str]
    user_actions: list[str]

    # 交接控制
    current_agent: str
    next: str
    handoff_reason: str

    # 解决状态
    resolved: bool
    resolution: str

# ========== 2. 定义代理 ==========

def tier1_support(state: SupportState):
    """L1 客服：初始接待和基础问题处理"""
    llm = ChatOpenAI(model="gpt-4o")

    messages = state["messages"]
    last_msg = messages[-1].content.lower()

    # 收集初步信息
    symptoms = []
    if "崩溃" in last_msg or "crash" in last_msg:
        symptoms.append("应用崩溃")
    if "慢" in last_msg or "slow" in last_msg:
        symptoms.append("性能缓慢")
    if "错误" in last_msg or "error" in last_msg:
        symptoms.append("错误提示")

    # 分类和优先级评估
    if "紧急" in last_msg or "无法使用" in last_msg:
        priority = "high"
        issue_type = "critical"
        next_agent = "tier2_technical"
        reason = "高优先级问题，需要技术支持"
        response = "这是紧急问题，我将立即为您转接技术支持团队（L2）"
    elif any(word in last_msg for word in ["技术", "bug", "崩溃", "错误"]):
        priority = "medium"
        issue_type = "technical"
        next_agent = "tier2_technical"
        reason = "技术问题，需要专业诊断"
        response = "这需要技术支持协助，我将为您转接技术团队（L2）"
    elif any(word in last_msg for word in ["如何", "怎么", "使用"]):
        priority = "low"
        issue_type = "general"
        next_agent = "end"
        reason = ""

        # L1 可以直接处理的常见问题
        prompt = f"""你是客服代理。用户问题：{last_msg}

        提供简洁的解决方案（< 100 字）。"""

        ai_response = llm.invoke([SystemMessage(content=prompt)])
        response = ai_response.content
    else:
        priority = "low"
        issue_type = "general"
        next_agent = "end"
        reason = ""
        response = "我来帮您解决这个问题"

    return {
        "messages": [AIMessage(content=f"L1 客服：{response}")],
        "symptoms": symptoms,
        "issue_type": issue_type,
        "priority": priority,
        "current_agent": "tier1",
        "next": next_agent,
        "handoff_reason": reason
    }

def tier2_technical(state: SupportState):
    """L2 技术支持：技术诊断和常规修复"""
    llm = ChatOpenAI(model="claude-sonnet-4-5")

    priority = state["priority"]
    symptoms = state.get("symptoms", [])
    messages = state["messages"]

    # 技术诊断
    diagnostic_prompt = f"""你是技术支持工程师。

    问题信息：
    - 优先级：{priority}
    - 症状：{', '.join(symptoms) if symptoms else '未知'}
    - 对话历史：{messages[-1].content}

    任务：
    1. 诊断问题
    2. 判断是否可以自行解决，还是需要升级给工程师（L3）

    如果问题涉及数据库、服务器、或需要代码修改，应升级。

    返回格式：
    诊断结果：[你的诊断]
    是否升级：[是/否]
    """

    diagnosis = llm.invoke([SystemMessage(content=diagnostic_prompt)])
    diagnosis_text = diagnosis.content

    # 判断是否升级
    should_escalate = "是否升级：是" in diagnosis_text or priority == "high"

    if should_escalate:
        # 收集更多诊断数据
        error_logs = ["Error: Database connection timeout", "Error: Service unavailable"]

        return {
            "messages": [AIMessage(content=f"L2 技术支持：{diagnosis_text}\n问题需要工程师深度介入，升级到 L3")],
            "error_logs": error_logs,
            "current_agent": "tier2",
            "next": "tier3_engineer",
            "handoff_reason": "需要代码级别排查和修复"
        }
    else:
        return {
            "messages": [AIMessage(content=f"L2 技术支持：{diagnosis_text}\n问题已解决")],
            "current_agent": "tier2",
            "next": "end",
            "resolved": True,
            "resolution": "L2 技术支持已解决"
        }

def tier3_engineer(state: SupportState):
    """L3 工程师：深度排查和代码修复"""
    llm = ChatOpenAI(model="claude-sonnet-4-5")

    symptoms = state.get("symptoms", [])
    error_logs = state.get("error_logs", [])

    # 工程师级别分析
    engineer_prompt = f"""你是资深工程师。

    问题信息：
    - 症状：{', '.join(symptoms)}
    - 错误日志：{', '.join(error_logs)}

    任务：
    1. 深度分析根本原因
    2. 提供代码级别的修复方案
    3. 给出预防措施
    """

    analysis = llm.invoke([SystemMessage(content=engineer_prompt)])

    return {
        "messages": [AIMessage(content=f"L3 工程师：{analysis.content}")],
        "current_agent": "tier3",
        "next": "end",
        "resolved": True,
        "resolution": "L3 工程师已深度修复"
    }

# ========== 3. 路由逻辑 ==========

def route(state: SupportState) -> Literal["tier1_support", "tier2_technical", "tier3_engineer", "end"]:
    """路由到下一个代理或结束"""
    next_agent = state.get("next", "tier1_support")
    if next_agent == "end":
        return END
    return next_agent

# ========== 4. 构建工作流 ==========

workflow = StateGraph(SupportState)

# 添加三层支持代理
workflow.add_node("tier1_support", tier1_support)
workflow.add_node("tier2_technical", tier2_technical)
workflow.add_node("tier3_engineer", tier3_engineer)

# 路由配置
workflow.add_edge(START, "tier1_support")
workflow.add_conditional_edges("tier1_support", route)
workflow.add_conditional_edges("tier2_technical", route)
workflow.add_conditional_edges("tier3_engineer", route)

# 编译
app = workflow.compile()

# ========== 5. 测试不同场景 ==========

test_scenarios = [
    {
        "name": "简单咨询（L1 处理）",
        "message": "如何重置密码？"
    },
    {
        "name": "技术问题（L1 → L2）",
        "message": "应用加载很慢，有时会卡住"
    },
    {
        "name": "严重故障（L1 → L2 → L3）",
        "message": "紧急！数据库连接失败，所有用户无法登录"
    }
]

for scenario in test_scenarios:
    print(f"\n{'='*60}")
    print(f"场景：{scenario['name']}")
    print(f"{'='*60}")

    result = app.invoke({
        "messages": [HumanMessage(content=scenario['message'])],
        "user_id": f"user_{uuid.uuid4().hex[:8]}",
        "session_id": f"session_{uuid.uuid4().hex[:8]}"
    })

    # 打印交接流程
    print(f"\n用户请求：{scenario['message']}")
    print(f"\n交接流程：")
    for msg in result["messages"]:
        if isinstance(msg, AIMessage):
            print(f"  {msg.content}")

    print(f"\n最终状态：")
    print(f"  - 问题类型：{result.get('issue_type', 'N/A')}")
    print(f"  - 优先级：{result.get('priority', 'N/A')}")
    print(f"  - 当前代理：{result.get('current_agent', 'N/A')}")
    print(f"  - 是否解决：{result.get('resolved', False)}")
    if result.get('handoff_reason'):
        print(f"  - 交接原因：{result['handoff_reason']}")

# 输出示例：
# ============================================================
# 场景：严重故障（L1 → L2 → L3）
# ============================================================
#
# 用户请求：紧急！数据库连接失败，所有用户无法登录
#
# 交接流程：
#   L1 客服：这是紧急问题，我将立即为您转接技术支持团队（L2）
#   L2 技术支持：[诊断结果...]
#   问题需要工程师深度介入，升级到 L3
#   L3 工程师：[深度分析和修复方案...]
#
# 最终状态：
#   - 问题类型：critical
#   - 优先级：high
#   - 当前代理：tier3
#   - 是否解决：True
#   - 交接原因：需要代码级别排查和修复

# 优势：
# - 自动化的多层级升级
# - 完整的状态传递（症状、日志、用户操作）
# - 每层代理专注于自己的职责范围
# - 灵活的交接决策（基于优先级、问题类型）
# - 生产级的错误处理和日志记录

"""
双向交接 - 支持升级和回退
"""
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END

class BidirectionalState(TypedDict):
    messages: Annotated[list, add_messages]
    current_tier: int  # 1, 2, 3
    next: str
    can_downgrade: bool  # 是否可以回退

def tier2_agent(state: BidirectionalState):
    """L2 代理：可能升级或回退"""
    messages = state["messages"]
    last_msg = messages[-1].content

    # 判断：问题是否实际上很简单
    if "其实没那么严重" in last_msg or "已经好了" in last_msg:
        return {
            "messages": [AIMessage(content="L2：问题已解决，无需进一步升级")],
            "next": "end",
            "can_downgrade": True
        }

    # 判断：是否需要升级
    if "还是不行" in last_msg or "更严重了" in last_msg:
        return {
            "messages": [AIMessage(content="L2：问题加剧，升级到 L3 工程师")],
            "current_tier": 3,
            "next": "tier3",
            "can_downgrade": False
        }

    return {
        "messages": [AIMessage(content="L2：正在处理...")],
        "next": "end"
    }

# 优势：
# - 灵活的流程控制
# - 避免不必要的升级
# - 支持问题降级（节省资源）

"""
条件触发交接 - 基于关键词、工具结果、时间
"""
from typing import TypedDict, Annotated
import datetime

class AutoHandoffState(TypedDict):
    messages: Annotated[list, add_messages]
    keywords_detected: list[str]
    tools_failed: bool
    elapsed_time: int  # 秒
    next: str

def smart_agent(state: AutoHandoffState):
    """智能代理：自动检测交接条件"""
    messages = state["messages"]
    last_msg = messages[-1].content.lower()

    # 条件 1：关键词触发
    critical_keywords = ["紧急", "数据丢失", "安全漏洞", "宕机"]
    detected = [kw for kw in critical_keywords if kw in last_msg]

    # 条件 2：工具失败触发
    tools_failed = state.get("tools_failed", False)

    # 条件 3：超时触发
    elapsed = state.get("elapsed_time", 0)
    timeout = elapsed > 300  # 5 分钟

    # 自动交接决策
    if detected or tools_failed or timeout:
        reason = []
        if detected:
            reason.append(f"检测到关键词：{', '.join(detected)}")
        if tools_failed:
            reason.append("工具调用失败")
        if timeout:
            reason.append("处理超时")

        return {
            "messages": [AIMessage(content=f"自动升级：{'; '.join(reason)}")],
            "keywords_detected": detected,
            "next": "specialist"
        }

    return {
        "messages": [AIMessage(content="正常处理中...")],
        "next": "end"
    }

# 优势：
# - 自动化决策
# - 多维度触发条件
# - 减少人工干预

# 交接日志记录
import logging

logger = logging.getLogger("handoffs")

def log_handoff(state: SupportState):
    """记录每次交接"""
    logger.info(f"""
    Handoff Event:
    - From: {state.get('current_agent')}
    - To: {state.get('next')}
    - Reason: {state.get('handoff_reason')}
    - User ID: {state.get('user_id')}
    - Session ID: {state.get('session_id')}
    - Timestamp: {datetime.now()}
    - Issue Type: {state.get('issue_type')}
    - Priority: {state.get('priority')}
    """)

# 在 LangSmith 中追踪交接
os.environ["LANGCHAIN_TRACING_V2"] = "true"

# 查看交接流程：
# - 每个代理的执行时间
# - 交接决策点
# - 状态变化

# 策略 1：限制交接次数
class SafeHandoffState(TypedDict):
    handoff_count: int
    max_handoffs: int  # 最大交接次数（如 3）

def safe_agent(state: SafeHandoffState):
    if state["handoff_count"] >= state["max_handoffs"]:
        return {"next": "end"}  # 强制结束

    # 正常交接逻辑
    return {
        "next": "other_agent",
        "handoff_count": state["handoff_count"] + 1
    }

# 策略 2：记录交接历史，避免重复
class HistoryState(TypedDict):
    handoff_history: list[str]  # ["agent1", "agent2"]

def check_cycle(state: HistoryState, target: str):
    """检查是否会形成循环"""
    if state["handoff_history"].count(target) >= 2:
        return True  # 同一个代理访问超过 2 次
    return False

# 使用 Checkpointer 保持会话
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()

app = workflow.compile(checkpointer=checkpointer)

# 调用时传递 thread_id
config = {"configurable": {"thread_id": "user123_session"}}

# 第一次调用（客服代理）
result1 = app.invoke(input1, config=config)

# 交接后的调用（技术支持代理）
# 会话状态自动保留
result2 = app.invoke(input2, config=config)