# CrewAI Sequential Pipeline 示例
from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="Senior Research Analyst",
    goal="Gather comprehensive data on {topic}",
    backstory="Expert at finding and synthesizing information",
    tools=[search_tool, scrape_tool]
)

writer = Agent(
    role="Technical Writer",
    goal="Transform research into clear article",
    backstory="Former tech journalist with 10 years experience"
)

editor = Agent(
    role="Editor",
    goal="Polish and fact-check the article",
    backstory="Award-winning editor at major publication"
)

research_task = Task(description="Research {topic}", agent=researcher)
write_task = Task(description="Write article based on research", agent=writer)
edit_task = Task(description="Edit and finalize article", agent=editor)

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

# LangGraph Fan-out/Fan-in 实现
from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Annotated
import operator

class State(TypedDict):
    topics: List[str]
    analyses: Annotated[list, operator.add]
    final_report: str

def scatter(state: State):
    """将大任务拆分为并行子任务"""
    return {"topics": ["market_analysis", "competitor_analysis", 
                       "tech_trends", "regulatory_landscape"]}

def analyze_topic(state: State):
    """每个Agent独立分析一个维度"""
    topic = state["current_topic"]
    result = analyst_agent.invoke({"task": topic})
    return {"analyses": [result]}

def fan_in(state: State):
    """聚合所有并行结果"""
    combined = synthesis_agent.invoke({
        "results": state["analyses"],
        "instruction": "综合所有分析，生成最终报告"
    })
    return {"final_report": combined}

# 构建并行图
graph = StateGraph(State)
graph.add_node("scatter", scatter)
graph.add_node("analyze_market", analyze_topic)
graph.add_node("analyze_competitor", analyze_topic)
graph.add_node("analyze_trends", analyze_topic)
graph.add_node("fan_in", fan_in)

graph.add_edge("scatter", "analyze_market")
graph.add_edge("scatter", "analyze_competitor")
graph.add_edge("scatter", "analyze_trends")
graph.add_edge("analyze_market", "fan_in")
graph.add_edge("analyze_competitor", "fan_in")
graph.add_edge("analyze_trends", "fan_in")

# LangGraph Supervisor 模式
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.types import Command

def supervisor(state: MessagesState) -> Command:
    """监督者：分析当前状态，决定下一步"""
    response = supervisor_llm.invoke([
        {"role": "system", "content": """你是任务协调者。
        根据对话历史决定下一步：
        - 需要研究 → 调用 research_agent
        - 需要计算 → 调用 math_agent  
        - 需要写代码 → 调用 code_agent
        - 所有任务完成 → 调用 FINISH
        """},
        *state["messages"]
    ])

    goto = response.tool_calls[0]["name"]
    return Command(goto=goto)

def research_agent(state: MessagesState):
    result = research_chain.invoke(state["messages"])
    return {"messages": [{"role": "assistant", "content": result}]}

# 构建Supervisor图
graph = StateGraph(MessagesState)
graph.add_node("supervisor", supervisor)
graph.add_node("research_agent", research_agent)
graph.add_node("math_agent", math_agent_node)
graph.add_node("code_agent", code_agent_node)

# 从supervisor到各个worker
graph.add_edge(START, "supervisor")
for agent in ["research_agent", "math_agent", "code_agent"]:
    graph.add_edge(agent, "supervisor")  # 每个worker完成后回到supervisor

graph.add_conditional_edges("supervisor", lambda x: x["goto"])

# LangChain Plan-and-Execute Agent
from langchain_experimental.plan_and_execute import (
    PlanAndExecute, Plan, PlanInput, Step
)

planner_prompt = """你是一个任务规划师。
给定一个目标，将其分解为清晰的执行步骤。

目标：{input}

输出格式：
1. [步骤描述]
2. [步骤描述]
...
"""

# Planner: 生成完整计划
planner = ChatOpenAI(model="gpt-4o").with_structured_output(Plan)

# Executor: 逐步执行
executor = AgentExecutor.from_agent_and_tools(
    agent=chat_agent,
    tools=[search_tool, calculator_tool, code_tool]
)

plan_and_execute = PlanAndExecute(
    planner=planner,
    executor=executor,
    planner_prompt=planner_prompt
)

# 执行
result = plan_and_execute.invoke({
    "input": "分析Tesla 2025年Q4财报，与行业平均对比，给出投资建议"
})
# Planner自动生成：
# Step 1: 搜索Tesla 2025 Q4财报数据
# Step 2: 提取关键财务指标（营收、利润率、现金流）
# Step 3: 搜索汽车行业2025年Q4平均指标
# Step 4: 计算Tesla与行业平均的差异
# Step 5: 综合分析并给出投资建议

# LangGraph 带动态重规划的 Plan-Execute
from langgraph.graph import StateGraph

class PlanExecuteState(TypedDict):
    input: str
    plan: List[str]
    past_steps: List[tuple]
    response: str

def plan_step(state: PlanExecuteState):
    """初始规划或重新规划"""
    if state.get("response") and "需要重规划" in state["response"]:
        # 基于已有执行结果重新规划
        plan = planner.invoke({
            "original_goal": state["input"],
            "completed_steps": state["past_steps"],
            "reason_for_replan": state["response"]
        })
    else:
        plan = planner.invoke({"input": state["input"]})
    return {"plan": plan.steps}

def execute_step(state: PlanExecuteState):
    task = state["plan"][0]
    result = executor.invoke({"task": task})
    return {
        "past_steps": [(task, result)] + state.get("past_steps", []),
        "plan": state["plan"][1:]
    }

def replan_or_finish(state: PlanExecuteState):
    """判断是否需要重规划"""
    if not state["plan"]:
        return "finish"
    check = replanner.invoke({
        "remaining_plan": state["plan"],
        "completed": state["past_steps"]
    })
    return "replan" if check.needs_replan else "execute"

# ROMA 风格递归分解（简化实现）
class RecursiveAgent:
    def __init__(self, task, depth=0, max_depth=5):
        self.task = task
        self.depth = depth
        self.max_depth = max_depth
        self.sub_agents = []

    def execute(self):
        # 尝试直接执行
        result = self._try_execute()

        if result.status == "success":
            return result.output

        if result.status == "too_complex":
            if self.depth >= self.max_depth:
                return self._fallback(result)

            # 递归分解：LLM生成子任务
            subtasks = self._decompose(result.reason)
            for subtask in subtasks:
                child = RecursiveAgent(
                    task=subtask,
                    depth=self.depth + 1,
                    max_depth=self.max_depth
                )
                self.sub_agents.append(child)

            # 并行执行子Agent
            sub_results = parallel_execute(self.sub_agents)

            # 合并子结果，重新尝试原始任务
            return self._merge_and_retry(sub_results)

    def _decompose(self, reason):
        """LLM根据失败原因递归分解"""
        response = llm.invoke([
            {"role": "system", "content": f"""任务'{self.task}'过于复杂。
            原因：{reason}
            将其拆分为2-4个独立的、更小的子任务。
            每个子任务应该可以被单独完成。"""},
        ])
        return parse_subtasks(response)

# 使用
root_agent = RecursiveAgent(
    task="构建一个完整的电商推荐系统，包括数据管道、模型训练和A/B测试框架",
    max_depth=4
)
result = root_agent.execute()

# GPT-HTN-Planner 简化架构
class HTNDomain:
    """HTN领域定义：任务分解规则"""

    methods = {
        "write_tech_report": [
            {  # Method 1: 标准报告流程
                "preconditions": {"has_topic": True, "depth": "standard"},
                "subtasks": ["research_topic", "create_outline", 
                           "write_sections", "review_and_edit"]
            },
            {  # Method 2: 快速报告流程
                "preconditions": {"has_topic": True, "depth": "quick"},
                "subtasks": ["quick_research", "write_brief_report"]
            }
        ],
        "research_topic": [
            {
                "preconditions": {"research_type": "academic"},
                "subtasks": ["search_papers", "extract_key_findings", 
                           "summarize_literature"]
            },
            {
                "preconditions": {"research_type": "market"},
                "subtasks": ["gather_market_data", "analyze_trends",
                           "competitive_landscape"]
            }
        ],
        # 原子任务（不再分解）
        "search_papers": None,  # 直接由Agent执行
        "extract_key_findings": None,
        "summarize_literature": None,
        # ...
    }

### 优缺点分析

HTN的核心竞争力在于**确定性**和**可解释性**。每一步分解都有明确的前提条件和规则，审计时可以完整回溯"为什么选择了这条分解路径"。这在金融合规、医疗决策等对可解释性有严格要求的场景中是关键优势。

HTN的局限性在于**领域知识的前期投入**。你需要为每个应用领域手工编写分解规则库——这本质上是在把领域专家的知识编码为可执行规则。2026年的趋势是用LLM自动生成HTN规则，然后人工审核和修正，将规则库的开发时间从数周缩短到数天。

### 生产表现

HTN在**高度结构化的垂直领域**中表现最佳。某医疗AI平台用HTN+LLM组合做"患者症状分析→检查建议→诊断推理→治疗方案推荐"的四层分解，准确率从纯LLM规划的78%提升到93%。关键在于HTN规则确保了分解路径始终符合临床指南——LLM不会因为创意发散而建议不合理的检查顺序。
## 六大范式架构全景图

```mermaid
graph TB
    subgraph 流水线分解
        A1[输入] --> A2[Agent-1: 研究]
        A2 --> A3[Agent-2: 撰写]
        A3 --> A4[Agent-3: 编辑]
        A4 --> A5[输出]
    end

    subgraph 散射-聚合分解
        B1[输入] --> B2[Scatter: 任务切分]
        B2 --> B3[Agent-A]
        B2 --> B4[Agent-B]
        B2 --> B5[Agent-C]
        B3 --> B6[Aggregate: 结果聚合]
        B4 --> B6
        B5 --> B6
        B6 --> B7[输出]
    end

    subgraph 层级监督分解
        C1[输入] --> C2[Supervisor Agent]
        C2 -->|分配| C3[研究Agent]
        C2 -->|分配| C4[计算Agent]
        C2 -->|分配| C5[编码Agent]
        C3 -->|汇报| C2
        C4 -->|汇报| C2
        C5 -->|汇报| C2
        C2 --> C6[输出]
    end

    subgraph 先规划后执行
        D1[输入] --> D2[Planner: 生成计划]
        D2 --> D3[Step 1]
        D3 --> D4[Step 2]
        D4 --> D5[Step 3]
        D5 --> D6{需要重规划?}
        D6 -->|是| D2
        D6 -->|否| D7[输出]
    end

    subgraph 递归自分解
        E1[Root Agent] -->|太复杂| E2[子Agent-1]
        E1 -->|太复杂| E3[子Agent-2]
        E2 -->|太复杂| E4[孙Agent-1]
        E2 -->|可执行| E5[结果]
        E3 -->|可执行| E6[结果]
        E4 -->|可执行| E7[结果]
    end

    subgraph HTN分解
        F1[Goal] --> F2{HTN Rule Match}
        F2 -->|Rule-1| F3[Method-A: 4个子任务]
        F2 -->|Rule-2| F4[Method-B: 2个子任务]
        F3 --> F5[原子任务集]
        F4 --> F6[原子任务集]
        F5 --> F7[DAG执行计划]
        F6 --> F7
    end

flowchart TD
    START[复杂任务] --> Q1{任务步骤间<br>有依赖关系?}

    Q1 -->|完全独立| Q2{子任务是否<br>同构?}
    Q1 -->|顺序依赖| Q3{依赖是否线性?}
    Q1 -->|复杂依赖| Q4{分解规则<br>可预定义?}

    Q2 -->|是| FANOUT[散射-聚合分解]
    Q2 -->|否| Q5{需要动态<br>分配?}

    Q3 -->|是| PIPELINE[流水线分解]
    Q3 -->|否| Q6{步骤数 > 5<br>且部分可预测?}

    Q4 -->|是| Q7{领域知识<br>充足?}
    Q4 -->|否| Q8{探索性任务?<br>分解路径未知?}

    Q5 -->|是| SUPERVISOR[层级监督分解]
    Q5 -->|否| FANOUT

    Q6 -->|是| PLANEXEC[先规划后执行]
    Q6 -->|否| SUPERVISOR

    Q7 -->|是| HTN[HTN分层任务网络]
    Q7 -->|否| PLANEXEC

    Q8 -->|是| RECURSIVE[递归自分解]
    Q8 -->|否| SUPERVISOR

    style FANOUT fill:#4CAF50,color:#fff
    style PIPELINE fill:#2196F3,color:#fff
    style SUPERVISOR fill:#FF9800,color:#fff
    style PLANEXEC fill:#9C27B0,color:#fff
    style HTN fill:#F44336,color:#fff
    style RECURSIVE fill:#607D8B,color:#fff

┌─────────────────────────────────────────────────────┐
│                   HTN Planner                        │
│  (定义任务分解规则、前置条件、合规约束)                  │
├─────────────────────────────────────────────────────┤
│                Supervisor Layer                       │
│  (动态分配、异常处理、重规划)                           │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐           │
│  │ Agent-A  │  │ Agent-B  │  │ Agent-C  │           │
│  │(Fan-out) │  │(Pipeline)│  │(ReAct)   │           │
│  │ ┌──┬──┐  │  │ A1→A2→A3 │  │ 循环推理  │           │
│  │ │A1│A2│  │  │          │  │          │           │
│  │ └──┴──┘  │  │          │  │          │           │
│  └──────────┘  └──────────┘  └──────────┘           │
└─────────────────────────────────────────────────────┘