第九章：跨框架对比——多 Agent 协作模式

from langgraph.graph import StateGraph, START, END

# 子图1：研究团队
research_graph = StateGraph(ResearchState)
research_graph.add_node("web_search", web_search_node)
research_graph.add_node("doc_search", doc_search_node)
research_graph.add_node("synthesize", synthesize_node)
research_graph.add_edge(START, "web_search")
research_graph.add_edge(START, "doc_search")  # 并行
research_graph.add_edge("web_search", "synthesize")
research_graph.add_edge("doc_search", "synthesize")
research_graph.add_edge("synthesize", END)

# 子图2：写作团队
writing_graph = StateGraph(WritingState)
writing_graph.add_node("draft", draft_node)
writing_graph.add_node("review", review_node)
writing_graph.add_node("revise", revise_node)
writing_graph.add_edge(START, "draft")
writing_graph.add_edge("draft", "review")
# 条件边：审核通过→结束，不通过→修改
writing_graph.add_conditional_edges("review", quality_check, {
    "pass": END,
    "fail": "revise"
})
writing_graph.add_edge("revise", "review")  # 修改后重新审核

# 主图：嵌套子图
main_graph = StateGraph(MainState)
main_graph.add_node("research_team", research_graph.compile())  # 子图作为节点
main_graph.add_node("writing_team", writing_graph.compile())
main_graph.add_edge(START, "research_team")
main_graph.add_edge("research_team", "writing_team")
main_graph.add_edge("writing_team", END)

def router(state: AgentState) -> str:
    """根据状态动态决定下一步"""
    last_message = state["messages"][-1]
    
    if "需要搜索" in last_message.content:
        return "search_agent"
    elif "需要代码" in last_message.content:
        return "coding_agent"
    elif "需要分析" in last_message.content:
        return "analysis_agent"
    else:
        return "final_answer"

graph.add_conditional_edges("dispatcher", router, {
    "search_agent": "search_agent",
    "coding_agent": "coding_agent",
    "analysis_agent": "analysis_agent",
    "final_answer": "final_answer",
})

# LangGraph 中实现 Coordinator 模式
def coordinator_node(state):
    """中心调度节点——决定下一步谁来"""
    # 读取当前状态，用 LLM 决策
    response = llm.invoke(f"""
    当前任务: {state['current_task']}
    已完成: {state['completed_steps']}
    请决定下一步由谁执行: research / code / review / done
    """)
    return {"next_agent": response.content}

def route_to_agent(state):
    return state["next_agent"]

graph.add_node("coordinator", coordinator_node)
graph.add_node("research", research_node)
graph.add_node("code", code_node)
graph.add_node("review", review_node)

graph.add_conditional_edges("coordinator", route_to_agent, {
    "research": "research",
    "code": "code",
    "review": "review",
    "done": END,
})

# 每个 Agent 执行完回到 Coordinator
graph.add_edge("research", "coordinator")
graph.add_edge("code", "coordinator")
graph.add_edge("review", "coordinator")

crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, write_task, edit_task],
    process=Process.sequential
)
# 严格按 tasks 列表顺序执行

crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[complex_task],
    process=Process.hierarchical,
    manager_llm=ChatOpenAI(model="gpt-4"),
    manager_agent=None  # 自动创建 Manager
)
# Manager Agent 接收任务后，选择子 Agent 分配
# 子 Agent 完成后汇报给 Manager
# Manager 综合判断是否完成

from crewai.flow.flow import Flow, start, listen, router

class DynamicFlow(Flow):
    @start()
    def classify_request(self):
        """分类用户请求"""
        return classifier_crew.kickoff()
    
    @router(classify_request)
    def route(self, classification):
        """根据分类结果路由"""
        if classification.category == "technical":
            return "tech_flow"
        elif classification.category == "creative":
            return "creative_flow"
        return "general_flow"
    
    @listen("tech_flow")
    def handle_tech(self):
        return tech_crew.kickoff()
    
    @listen("creative_flow")
    def handle_creative(self):
        return creative_crew.kickoff()
    
    @listen("general_flow")
    def handle_general(self):
        return general_crew.kickoff()

from agno.team import Team

# 1. Coordinate（协调模式 = 星型）
team = Team(
    agents=[researcher, writer, editor],
    mode="coordinate",
    # Leader 协调成员工作，综合结果
)

# 2. Route（路由模式 = 条件分发）
team = Team(
    agents=[tech_agent, creative_agent, data_agent],
    mode="route",
    # 根据输入自动路由到最合适的 Agent
)

# 3. Collaborate（协作模式 = 网状）
team = Team(
    agents=[analyst, designer, developer],
    mode="collaborate",
    # 所有 Agent 都看到输入，各自贡献，最后合并
)

# 4. Broadcast（广播模式 = 并行独立）
team = Team(
    agents=[reviewer_1, reviewer_2, reviewer_3],
    mode="broadcast",
    # 同一个输入发给所有 Agent，收集所有回复
    # 适合"多审稿人独立评审"场景
)

# 节点不"调用"其他节点——它们通过共享 State 间接通信
def research_node(state):
    results = search(state["query"])
    return {"findings": results}  # 写入 State

def analyze_node(state):
    data = state["findings"]  # 从 State 读取前一个节点的输出
    analysis = analyze(data)
    return {"analysis": analysis}

# 上游 Task 的输出文本自动注入下游 Task 的上下文
research_task = Task(description="研究X", agent=researcher)
writing_task = Task(description="基于研究写文章", context=[research_task], agent=writer)
# writer 的 prompt 中会出现："前置任务输出: [research结果]"

# OpenAI Swarm 的设计哲学（Python 参考实现）
class Agent:
    name: str
    instructions: str
    functions: list  # 包括 transfer_to_xxx 函数

def transfer_to_support():
    """Transfer the conversation to the support agent."""
    return support_agent  # 返回 Agent 对象 = 执行交接

# 特点：
# 1. 轻量——没有框架级的"编排器"
# 2. Agent 自治——自己决定交给谁
# 3. 对话历史 = 状态——交接就是"换一个 Agent 来读同一段历史"
# 4. 无持久化——纯内存，适合单次对话

# 在 LangGraph 中，"交接"就是图的边
# Agent A 执行完 → 沿着边 → Agent B 开始执行

# 子图模式的"交接"：
def should_handoff(state):
    """条件边：决定是否交接给子图"""
    if state["needs_specialist"]:
        return "specialist_subgraph"
    return "continue"

graph.add_conditional_edges("generalist", should_handoff, {
    "specialist_subgraph": "specialist",
    "continue": "generalist_next_step",
})

# 子图执行完毕后，控制权自动返回父图
# 这种"交接"是结构化的——由图的拓扑决定，不是 Agent 自主决定

# Manager Agent 决定把任务委派给谁
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[complex_task],
    process=Process.hierarchical,
    manager_llm=ChatOpenAI(model="gpt-4")
)

# 内部执行：
# 1. Manager 分析 complex_task
# 2. Manager 决定："先让 researcher 去收集信息"
# 3. researcher 执行，结果返回给 Manager
# 4. Manager 决定："现在让 writer 写文章"
# 5. writer 执行，结果返回给 Manager
# 6. Manager 判断是否完成

# 这不是"交接"——是"委派+回收"
# 控制权始终在 Manager 手里，子 Agent 只是"被调用"

from langgraph.graph import StateGraph
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import interrupt

# 方式1：interrupt() 函数——在任何地方暂停等待人类输入
def sensitive_action_node(state):
    """执行敏感操作前，暂停等待人类确认"""
    action = state["pending_action"]
    
    # 暂停执行，等待人类确认
    human_response = interrupt(
        f"即将执行: {action}\n请确认(yes/no/modify): "
    )
    
    if human_response == "no":
        return {"action_result": "用户取消了操作"}
    elif human_response.startswith("modify:"):
        action = human_response.replace("modify:", "")
    
    result = execute_action(action)
    return {"action_result": result}

# 方式2：编译时声明"在某些节点前中断"
app = graph.compile(
    checkpointer=SqliteSaver.from_conn_string("db"),
    interrupt_before=["dangerous_node"],  # 在这些节点执行前暂停
    interrupt_after=["review_node"],       # 在这些节点执行后暂停
)

# 使用流程：
config = {"configurable": {"thread_id": "task-001"}}

# 执行到中断点会暂停
result = app.invoke(input_data, config)
# result.next = ["dangerous_node"]  ← 告诉你"接下来要执行这个，等你确认"

# 人类审查 State，可以修改后再继续
current_state = app.get_state(config)
print(current_state.values)  # 看当前状态

# 方式A：继续执行（确认）
app.invoke(None, config)  # 传 None = "继续"

# 方式B：修改 State 后继续
app.update_state(config, {"pending_action": "修改后的操作"})
app.invoke(None, config)

# 方式C：跳过这个节点
app.update_state(config, {"skip_dangerous": True}, as_node="dangerous_node")

from crewai.flow.flow import Flow, start, listen, human_input

class ReviewFlow(Flow):
    @start()
    def generate_content(self):
        return content_crew.kickoff()
    
    @human_input(generate_content)
    def review_content(self, content, human_feedback):
        """
        generate_content 执行后暂停等待人类反馈
        human_feedback 是人类输入的文本
        """
        if "approved" in human_feedback.lower():
            return content
        else:
            # 把人类反馈传给修改 crew
            return revision_crew.kickoff(
                inputs={"original": content, "feedback": human_feedback}
            )

from langgraph.constants import Send
from langgraph.graph import StateGraph

class State(TypedDict):
    documents: list[str]  # 输入的文档列表
    summaries: Annotated[list[str], operator.add]  # 汇总结果

def fan_out(state):
    """根据文档数量动态创建并行分支"""
    return [
        Send("summarize", {"doc": doc, "index": i})
        for i, doc in enumerate(state["documents"])
    ]
    # 如果有 5 个文档，就并行启动 5 个 summarize 节点实例

def summarize_node(state):
    """每个实例处理一个文档"""
    summary = llm.summarize(state["doc"])
    return {"summaries": [summary]}  # reducer 会合并所有实例的结果

graph = StateGraph(State)
graph.add_node("summarize", summarize_node)
graph.add_conditional_edges("start", fan_out)
graph.add_edge("summarize", "aggregate")

# CrewAI 可以在运行时构造 Task 列表
def create_review_crew(chapters: list[str]) -> Crew:
    """根据章节数量动态创建 Task"""
    tasks = []
    for i, chapter in enumerate(chapters):
        task = Task(
            description=f"审核第{i+1}章: {chapter[:100]}...",
            agent=reviewer,
            expected_output="审核意见",
        )
        tasks.append(task)
    
    # 最后汇总
    summary_task = Task(
        description="汇总所有章节审核意见",
        context=tasks,  # 依赖所有审核 Task
        agent=editor,
        expected_output="综合审核报告",
    )
    tasks.append(summary_task)
    
    return Crew(agents=[reviewer, editor], tasks=tasks, process=Process.sequential)

from langgraph.graph import StateGraph
from langgraph.pregel import RetryPolicy

# 节点级重试策略
graph.add_node("risky_node", risky_function, retry=RetryPolicy(
    max_attempts=3,
    initial_interval=1.0,    # 初始等待1秒
    backoff_factor=2.0,      # 指数退避
    retry_on=(TimeoutError, APIError),  # 只对这些错误重试
))

# 结合 Checkpoint 的恢复：
# 1. 节点失败 → 先按 retry_policy 重试
# 2. 重试全部失败 → 抛出异常
# 3. 但 Checkpoint 已保存了最后一个成功节点的状态
# 4. 修复问题后，从 Checkpoint 恢复继续

# 高级：自定义错误处理节点
def error_handler(state):
    """当某个节点失败时，走到这个错误处理节点"""
    error = state["last_error"]
    if "rate_limit" in str(error):
        return {"action": "wait_and_retry", "wait_seconds": 60}
    elif "invalid_format" in str(error):
        return {"action": "retry_with_different_prompt"}
    return {"action": "abort"}

graph.add_node("error_handler", error_handler)
# 可以用条件边把"失败状态"路由到 error_handler

# CrewAI 的重试相对简单
from crewai import Agent

agent = Agent(
    role="Researcher",
    max_retry_limit=3,          # 最多重试3次
    max_execution_time=300,     # 超时限制（秒）
    allow_delegation=True,      # 允许委派给其他Agent（如果自己搞不定）
)

# Task 级别
task = Task(
    description="...",
    agent=agent,
    max_retries=2,  # Task 级重试
)

# 没有：
# - 指数退避
# - 熔断机制
# - 自定义重试条件
# - 从中间状态恢复