stateDiagram-v2
    [*] --> 创建: Crew(...) 或 @crew 装饰器
    创建 --> 配置: 设置 agents/tasks/process
    配置 --> 就绪: 等待 kickoff
    就绪 --> 回调前: @before_kickoff
    回调前 --> 插值: 输入参数插值
    插值 --> 进程选择: sequential 或 hierarchical

    进程选择 --> Sequential: process=sequential
    进程选择 --> Hierarchical: process=hierarchical

    Sequential --> 任务执行: _execute_tasks()
    Hierarchical --> 创建管理者: _create_manager_agent()
    创建管理者 --> 任务执行

    任务执行 --> 同步任务: 逐个执行
    任务执行 --> 异步任务: 后台执行

    同步任务 --> 条件任务: ConditionalTask 判断
    异步任务 --> 等待完成: future.result()

    条件任务 --> 聚合输出: CrewOutput
    等待完成 --> 聚合输出

    聚合输出 --> 回调后: @after_kickoff
    回调后 --> 完成: return CrewOutput
    完成 --> [*]

flowchart TB
    subgraph "上游接口层 (Entry Points)"
        UserCode[用户代码<br/>crew.kickoff()]
        Flow[Flow 引擎<br/>调用 Crew]
        CLI[CLI 工具<br/>crewai run]
        API[API 接口<br/>HTTP/RPC]
    end

    subgraph "Crew 核心协调层"
        Crew[Crew 编排器<br/>kickoff/kickoff_async]
        CrewOutput[CrewOutput<br/>输出聚合]
        Callbacks[生命周期回调<br/>before/after_kickoff]
    end

    subgraph "流程选择层 (Process Layer)"
        ProcessRouter{Process 类型}
        Sequential[Sequential Process<br/>顺序执行流程]
        Hierarchical[Hierarchical Process<br/>层级管理流程]
    end

    subgraph "任务执行层 (Task Execution)"
        ExecuteTasks[_execute_tasks<br/>任务循环执行器]
        GetAgent[_get_agent_to_use<br/>Agent 选择]
        PrepareTools[_prepare_tools<br/>工具准备]
        GetContext[_get_context<br/>上下文聚合]
        ConditionalCheck[_handle_conditional_task<br/>条件判断]
    end

    subgraph "Manager 管理层 (Hierarchical Only)"
        ManagerAgent[Manager Agent<br/>自动生成协调者]
        DelegateTools[委派工具<br/>DelegateTaskTool]
        AskTools[询问工具<br/>AskQuestionTool]
    end

    subgraph "Task 执行层"
        TaskSync[Task.execute_sync<br/>同步执行]
        TaskAsync[Task.execute_async<br/>异步执行]
        FutureManager[Future 管理<br/>异步任务协调]
    end

    subgraph "Agent 执行层"
        AgentExecutor[Agent.execute_task<br/>任务执行]
        AgentExecutorImpl[AgentExecutor.invoke<br/>LLM 调用]
        ToolInvoke[Tool 调用<br/>工具执行]
        LLMCall[LLM 调用<br/>生成响应]
    end

    subgraph "资源管理层 (Shared Resources)"
        Memory[Memory 系统<br/>Short/Long/Entity/External]
        Cache[Cache 系统<br/>工具结果缓存]
        Knowledge[Knowledge 系统<br/>RAG 知识库]
        RPMController[RPM Controller<br/>速率限制]
    end

    subgraph "事件与日志层 (Observability)"
        EventBus[Event Bus<br/>事件总线]
        Logger[Logger<br/>日志系统]
        Storage[Task Output Storage<br/>执行记录持久化]
        Tracing[Tracing<br/>OpenTelemetry]
    end

    subgraph "输出处理层"
        TaskOutput[TaskOutput<br/>任务输出]
        Guardrails[Guardrails<br/>输出验证]
        OutputFormat[Output Format<br/>Pydantic/JSON]
    end

    %% 主流程连接
    UserCode --> Crew
    Flow --> Crew
    CLI --> Crew
    API --> Crew

    Crew --> Callbacks
    Callbacks --> ProcessRouter

    ProcessRouter -->|sequential| Sequential
    ProcessRouter -->|hierarchical| Hierarchical

    Sequential --> ExecuteTasks
    Hierarchical --> ManagerAgent
    ManagerAgent --> DelegateTools
    ManagerAgent --> AskTools
    Hierarchical --> ExecuteTasks

    ExecuteTasks --> GetAgent
    ExecuteTasks --> PrepareTools
    ExecuteTasks --> GetContext
    ExecuteTasks --> ConditionalCheck

    ExecuteTasks --> TaskSync
    ExecuteTasks --> TaskAsync
    TaskAsync --> FutureManager

    TaskSync --> AgentExecutor
    FutureManager --> AgentExecutor

    AgentExecutor --> AgentExecutorImpl
    AgentExecutorImpl --> ToolInvoke
    AgentExecutorImpl --> LLMCall

    AgentExecutor --> TaskOutput
    TaskOutput --> Guardrails
    Guardrails --> OutputFormat
    OutputFormat --> CrewOutput

    %% 资源管理连接
    Crew -.->|初始化| Memory
    Crew -.->|初始化| Cache
    Crew -.->|初始化| Knowledge
    Crew -.->|初始化| RPMController

    AgentExecutor -.->|查询| Memory
    AgentExecutor -.->|查询| Knowledge
    ToolInvoke -.->|缓存| Cache
    LLMCall -.->|限速| RPMController

    %% 事件日志连接
    Crew -->|发送| EventBus
    ExecuteTasks -->|发送| EventBus
    AgentExecutor -->|发送| EventBus
    Crew -->|记录| Logger
    ExecuteTasks -->|存储| Storage
    Crew -.->|追踪| Tracing

    style Crew fill:#e1bee7
    style ProcessRouter fill:#fff9c4
    style ManagerAgent fill:#c5e1a5
    style AgentExecutor fill:#b3e5fc
    style Memory fill:#ffccbc
    style EventBus fill:#d1c4e9

Agent.execute_task()
  → Agent.create_agent_executor()  # 如果未初始化
  → AgentExecutor.invoke()         # LangChain executor
    → LLM 调用（生成响应）
    → Tool 调用（如果需要）
    → 循环直到完成
  → 返回最终输出

classDiagram
    class Crew {
        +UUID id
        +list[BaseAgent] agents
        +list[Task] tasks
        +Process process
        +bool verbose
        +bool memory
        +ShortTermMemory short_term_memory
        +LongTermMemory long_term_memory
        +EntityMemory entity_memory
        +ExternalMemory external_memory
        +bool cache
        +Cache cache_handler
        +Knowledge knowledge
        +int max_rpm
        +RPMController rpm_controller
        +str prompt_file
        +I18N i18n
        +BaseLLM manager_llm
        +str manager_agent
        +bool step_callback
        +bool task_callback
        +bool full_output
        +bool share_crew
        +dict inputs
        +bool respect_context_window
        +bool planning
        +list[Task] planning_tasks
        +EmbedderConfig embedder
        +bool training
        +dict training_data
        +kickoff(inputs)
        +kickoff_for_each(inputs)
        +train(n_iterations, filename, inputs)
        +replay(task_id, inputs)
        +test(n_iterations, openai_model_name)
        +copy()
        +reset_memories(command_type)
        -_execute_tasks(tasks, start_index)
        -_run_sequential_process()
        -_run_hierarchical_process()
        -_get_context(task, task_outputs)
        -_prepare_tools(agent, task, tools)
        -_create_manager_agent()
        -_handle_conditional_task()
    }

    class CrewOutput {
        +str raw
        +BaseModel pydantic
        +dict json_dict
        +list[TaskOutput] tasks_output
        +UsageMetrics token_usage
        +json()
        +to_dict()
    }

    class Process {
        <<enumeration>>
        sequential
        hierarchical
    }

    class Memory {
        <<interface>>
        +save(value, metadata)
        +search(query, limit, score_threshold)
        +set_crew(crew)
    }

    class ShortTermMemory {
        +save(value, metadata)
        +search(query, limit, score_threshold)
    }

    class LongTermMemory {
        +save(value, metadata)
        +search(query, limit, score_threshold)
    }

    class EntityMemory {
        +save(value, metadata)
        +search(query, limit, score_threshold)
    }

    class ExternalMemory {
        +save(value, metadata)
        +search(query, limit, score_threshold)
    }

    class Cache {
        +read(tool, input)
        +add(tool, input, output)
    }

    class RPMController {
        +int max_rpm
        +int _current_rpm
        +check_or_wait()
        +stop_rpm_counter()
    }

    class I18N {
        +str prompt_file
        +dict _prompts
        +slice(slice)
        +errors(error)
        +tools(tool)
    }

    Crew --> CrewOutput : 生成
    Crew --> Process : 使用
    Crew --> Memory : 管理
    Memory <|-- ShortTermMemory : 实现
    Memory <|-- LongTermMemory : 实现
    Memory <|-- EntityMemory : 实现
    Memory <|-- ExternalMemory : 实现
    Crew --> Cache : 使用
    Crew --> RPMController : 使用
    Crew --> I18N : 使用

class UsageMetrics(BaseModel):
    total_tokens: int = 0
    prompt_tokens: int = 0
    completion_tokens: int = 0
    successful_requests: int = 0

class KickoffParams:
    inputs: Optional[dict[str, Any]] = None  # 输入参数字典

inputs = {
    "topic": "AI Safety",             # 模板变量
    "count": 5,                        # 模板变量
    "crew_chat_messages": [            # 会话历史
        {"role": "user", "content": "Find latest AI news"},
        {"role": "assistant", "content": "Here are the findings..."}
    ],
    "crewai_trigger_payload": {...}   # Trigger payload（特殊用途）
}

class CrewOutput(BaseModel):
    raw: str                         # 最后一个任务的原始输出
    pydantic: Optional[BaseModel]    # 最后一个任务的 Pydantic 输出
    json_dict: Optional[dict]        # 最后一个任务的 JSON 输出
    tasks_output: list[TaskOutput]   # 所有任务的输出列表
    token_usage: UsageMetrics        # Token 使用统计

def kickoff(self, inputs: dict[str, Any] | None = None) -> CrewOutput:
    """启动 Crew 执行"""
    self.inputs = inputs or {}

    try:
        # 1. 初始化（首次执行）
        if not self._execution_started:
            self._execution_started = True
            self._execution_logs = []
            self._task_outputs_replayed_from_memory = []

        # 2. 重置内部状态
        self._finished_tasks = set()
        self._tasks_index_map = {str(task.id): i for i, task in enumerate(self.tasks)}
        self._copy_agents()

        # 3. 执行 @before_kickoff 回调
        if hasattr(self.__class__, "before_kickoff"):
            self.__class__.before_kickoff(self)

        # 4. 发送 CrewStartedEvent
        crewai_event_bus.emit(
            self, CrewStartedEvent(inputs=self.inputs, crew=self)
        )

        # 5. 插值任务
        self._interpolate_tasks(self.inputs)

        # 6. 设置 token 处理
        self._set_tokens_process()

        # 7. 初始化内存和知识库
        self._initialize_memory()
        self._initialize_crew_knowledge()

        # 8. 注入关系（crew → agents → tasks）
        self._inject_relationships()

        # 9. 生成规划任务（如果启用）
        if self.planning:
            self.planning_tasks = self._create_planning_tasks()

        # 10. 选择执行流程
        if self.process == Process.sequential:
            result = self._run_sequential_process()
        else:
            result = self._run_hierarchical_process()

        # 11. 聚合 token 使用
        metrics = self._finish_execution(result.tasks_output)
        result.token_usage = metrics

        # 12. 执行 @after_kickoff 回调
        if hasattr(self.__class__, "after_kickoff"):
            self.__class__.after_kickoff(self, result)

        # 13. 发送 CrewCompletedEvent
        crewai_event_bus.emit(
            self, CrewCompletedEvent(output=result, crew=self)
        )

        return result

    except Exception as e:
        # 发送 CrewFailedEvent
        crewai_event_bus.emit(
            self, CrewFailedEvent(error=str(e), crew=self)
        )
        raise e

from crewai import Crew, Agent, Task

# 定义 Crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
    memory=True,
)

# 启动执行
result = crew.kickoff(inputs={"topic": "AI Safety"})

# 访问结果
print(result.raw)
print(result.token_usage.total_tokens)
for task_output in result.tasks_output:
    print(f"{task_output.description}: {task_output.raw}")

class MyFlow(Flow):
    @start()
    def run_crew_step(self):
        crew = MyCrew()
        result = crew.crew().kickoff(inputs={"query": self.state.query})
        self.state.crew_result = result.raw
        return result

sequenceDiagram
    autonumber
    participant User as 用户代码
    participant Crew as Crew 编排器
    participant CB as 回调系统
    participant Memory as Memory 系统
    participant Cache as Cache 系统
    participant Knowledge as Knowledge 系统
    participant Process as Process 层
    participant TaskExec as 任务执行层
    participant Task as Task
    participant Agent as Agent
    participant AgentExec as AgentExecutor
    participant LLM as LLM
    participant Tool as Tool
    participant Event as Event Bus
    participant Storage as Storage
    participant Output as Output 层

    %% 启动阶段
    User->>Crew: kickoff(inputs)
    activate Crew

    Crew->>Event: emit(CrewKickoffStartedEvent)
    Crew->>CB: 执行 before_kickoff_callbacks
    CB-->>Crew: 修改后的 inputs

    %% 初始化阶段
    Crew->>Crew: _interpolate_inputs(inputs)
    Note over Crew: 插值任务和 Agent 的模板变量

    Crew->>Memory: _initialize_memory()
    activate Memory
    Memory->>Memory: 创建 ShortTermMemory
    Memory->>Memory: 创建 LongTermMemory
    Memory->>Memory: 创建 EntityMemory
    Memory->>Memory: 创建 ExternalMemory (如果配置)
    Memory-->>Crew: Memory 系统就绪
    deactivate Memory

    Crew->>Cache: 初始化 CacheHandler
    Cache-->>Crew: Cache 就绪

    Crew->>Knowledge: _initialize_crew_knowledge()
    Knowledge-->>Crew: Knowledge 就绪

    Crew->>Crew: 为 Agent 注入资源
    Note over Crew: agent.crew = self<br/>agent.set_knowledge()<br/>agent.create_agent_executor()

    %% 流程选择
    alt process == sequential
        Crew->>Process: _run_sequential_process()
    else process == hierarchical
        Crew->>Process: _run_hierarchical_process()
        Process->>Process: _create_manager_agent()
        Note over Process: 创建 Manager Agent<br/>添加 DelegateTaskTool
    end

    %% 任务执行循环
    Process->>TaskExec: _execute_tasks(tasks)
    activate TaskExec

    loop 遍历每个任务
        TaskExec->>TaskExec: _get_agent_to_use(task)
        Note over TaskExec: Sequential: task.agent<br/>Hierarchical: manager_agent

        TaskExec->>TaskExec: _prepare_tools(agent, task)
        Note over TaskExec: 合并 agent.tools + task.tools<br/>添加缓存包装<br/>添加委派工具（如果允许）

        TaskExec->>TaskExec: _get_context(task, task_outputs)
        Note over TaskExec: 聚合前序任务的输出

        alt ConditionalTask
            TaskExec->>Task: should_execute(previous_output)
            alt 条件不满足
                Task-->>TaskExec: get_skipped_task_output()
                TaskExec->>Storage: _store_execution_log()
                Note over TaskExec: 跳过任务，继续下一个
            end
        end

        alt async_execution == True
            TaskExec->>Task: execute_async(agent, context, tools)
            activate Task
            Task->>Task: 创建 Future
            Task->>Agent: 后台线程执行
            Task-->>TaskExec: 返回 Future
            deactivate Task
            Note over TaskExec: 不等待，继续下一个任务
        else 同步任务
            alt 存在未完成的异步任务
                TaskExec->>TaskExec: _process_async_tasks(futures)
                Note over TaskExec: 等待所有异步任务完成<br/>收集结果
            end

            TaskExec->>Task: execute_sync(agent, context, tools)
            activate Task

            Task->>Event: emit(TaskStartedEvent)
            Task->>Agent: execute_task(task, context, tools)
            activate Agent

            %% Agent 执行
            Agent->>Agent: 构建 task_prompt
            Note over Agent: 包含：role, goal, backstory,<br/>task description, context

            alt 有 memory 系统
                Agent->>Memory: search(task.description)
                Memory-->>Agent: 相关记忆列表
                Note over Agent: 将记忆添加到 context
            end

            alt 有 knowledge 系统
                Agent->>Knowledge: query(task.description)
                Knowledge-->>Agent: 相关知识片段
                Note over Agent: 将知识添加到 context
            end

            Agent->>AgentExec: invoke(task_prompt)
            activate AgentExec

            loop LLM 循环（直到完成）
                AgentExec->>LLM: generate(prompt, tools)
                activate LLM
                LLM-->>AgentExec: LLM Response (text or tool call)
                deactivate LLM

                alt 需要调用工具
                    AgentExec->>Cache: read(tool, input)
                    alt 缓存命中
                        Cache-->>AgentExec: 缓存结果
                    else 缓存未命中
                        AgentExec->>Tool: invoke(input)
                        activate Tool
                        Tool-->>AgentExec: Tool Output
                        deactivate Tool
                        AgentExec->>Cache: add(tool, input, output)
                    end

                    AgentExec->>Memory: save(tool_output)
                    Note over AgentExec: 继续 LLM 循环
                else 任务完成
                    Note over AgentExec: 退出循环
                end
            end

            AgentExec-->>Agent: 最终输出
            deactivate AgentExec

            Agent->>Memory: save(final_output)
            Agent-->>Task: 最终输出字符串
            deactivate Agent

            %% 输出处理
            Task->>Output: _export_output(result)
            activate Output
            Output->>Output: 提取 pydantic_output
            Output->>Output: 提取 json_output
            Output-->>Task: (pydantic, json)
            deactivate Output

            Task->>Output: 创建 TaskOutput
            Output->>Output: TaskOutput(raw, pydantic, json_dict)

            alt 配置了 Guardrails
                loop 遍历每个 guardrail
                    Output->>Output: _invoke_guardrail_function()
                    alt 验证失败
                        Output->>Agent: 重新执行任务
                        Note over Output: 使用 guardrail 反馈<br/>重试直到成功或达到 max_retries
                    end
                end
            end

            Task->>Event: emit(TaskCompletedEvent)
            Task-->>TaskExec: TaskOutput
            deactivate Task

            TaskExec->>TaskExec: _process_task_result(task, output)
            TaskExec->>Storage: _store_execution_log(task, output)
            Note over Storage: 持久化任务输出<br/>支持回放功能
        end
    end

    alt 仍有未完成的异步任务
        TaskExec->>TaskExec: _process_async_tasks(futures)
    end

    TaskExec->>TaskExec: _create_crew_output(task_outputs)
    Note over TaskExec: 聚合所有任务输出<br/>选择最后一个有效输出作为主输出

    TaskExec-->>Crew: CrewOutput
    deactivate TaskExec

    %% 完成阶段
    Crew->>Crew: calculate_usage_metrics()
    Note over Crew: 统计所有 Agent 的 token 使用

    Crew->>CB: 执行 after_kickoff_callbacks
    CB-->>Crew: 修改后的 CrewOutput

    Crew->>Event: emit(CrewKickoffCompletedEvent)
    Crew-->>User: CrewOutput
    deactivate Crew

def kickoff_for_each(self, inputs: list[dict[str, Any]]) -> list[CrewOutput]:
    """批量执行 Crew"""
    results: list[CrewOutput] = []

    for input_data in inputs:
        # 重置执行状态
        self._execution_started = False

        # 执行单次 kickoff
        result = self.kickoff(inputs=input_data)
        results.append(result)

        # 清理内存（可选）
        if self.memory:
            self.reset_memories("long_term")

    return results

# 定义多组输入
inputs_list = [
    {"topic": "AI Safety", "count": 5},
    {"topic": "Climate Change", "count": 10},
    {"topic": "Healthcare", "count": 7},
]

# 批量执行
results = crew.kickoff_for_each(inputs=inputs_list)

# 处理结果
for i, result in enumerate(results):
    print(f"Result {i+1}:")
    print(result.raw)
    print(f"Tokens used: {result.token_usage.total_tokens}\n")

def train(
    self, n_iterations: int, filename: str, inputs: dict[str, Any] | None = None
) -> None:
    """训练 Crew"""
    # 1. 启用训练模式
    self.training = True

    # 2. 禁用内存（训练时不需要）
    original_memory_config = self.memory
    self.memory = False

    # 3. 初始化训练数据
    training_data_file = f"{filename}.pkl"
    if os.path.exists(training_data_file):
        with open(training_data_file, "rb") as f:
            self.training_data = pickle.load(f)
    else:
        self.training_data = {}

    # 4. 执行多轮训练
    for iteration in range(n_iterations):
        print(f"\n=== Training Iteration {iteration + 1}/{n_iterations} ===\n")

        # 执行 Crew
        result = self.kickoff(inputs=inputs)

        # 收集人工反馈
        print("\nCrew execution completed. Please provide feedback:")
        feedback = input("Was the output satisfactory? (yes/no): ")

        if feedback.lower() == "yes":
            # 保存成功的执行记录
            for task in self.tasks:
                if task.output:
                    self.training_data[str(task.id)] = {
                        "description": task.description,
                        "expected_output": task.expected_output,
                        "actual_output": task.output.raw,
                        "feedback": "positive",
                    }
        else:
            # 收集改进建议
            improvement = input("What should be improved? ")
            for task in self.tasks:
                if task.output:
                    self.training_data[str(task.id)] = {
                        "description": task.description,
                        "expected_output": task.expected_output,
                        "actual_output": task.output.raw,
                        "feedback": "negative",
                        "improvement": improvement,
                    }

        # 保存训练数据
        with open(training_data_file, "wb") as f:
            pickle.dump(self.training_data, f)

    # 5. 恢复配置
    self.training = False
    self.memory = original_memory_config

    print(f"\nTraining completed. Data saved to {training_data_file}")

# 训练 Crew
crew.train(
    n_iterations=5,
    filename="research_crew_training",
    inputs={"topic": "AI Trends"}
)

# 训练后，Crew 会利用训练数据优化 Agent 行为
# 训练数据保存在 research_crew_training.pkl

def replay(self, task_id: str, inputs: dict[str, Any] | None = None) -> CrewOutput:
    """从指定任务回放执行"""
    # 1. 查找任务索引
    if task_id not in self._tasks_index_map:
        raise ValueError(f"Task with id {task_id} not found")

    start_index = self._tasks_index_map[task_id]

    # 2. 加载历史执行记录
    self._load_execution_logs()

    # 3. 恢复前序任务的输出
    task_outputs_to_replay: list[TaskOutput] = []
    for i in range(start_index):
        task = self.tasks[i]
        # 从存储中读取任务输出
        saved_output = self._read_task_output_from_log(str(task.id))
        if saved_output:
            task_outputs_to_replay.append(saved_output)

    self._task_outputs_replayed_from_memory = task_outputs_to_replay

    # 4. 从指定任务开始执行
    if self.process == Process.sequential:
        return self._run_sequential_process_from_index(start_index)
    else:
        # Hierarchical 不支持 replay
        raise ValueError("Replay is not supported for hierarchical process")

# 首次执行
result = crew.kickoff(inputs={"topic": "AI Safety"})

# 获取某个任务的 ID
research_task_id = str(crew.tasks[0].id)

# 从该任务重新执行（例如修改了 Agent 配置后）
replay_result = crew.replay(
    task_id=research_task_id,
    inputs={"topic": "AI Safety"}
)

def test(
    self,
    n_iterations: int,
    openai_model_name: str,
    inputs: dict[str, Any] | None = None,
) -> None:
    """测试 Crew 性能"""
    results = []

    for i in range(n_iterations):
        print(f"\n=== Test Iteration {i + 1}/{n_iterations} ===\n")

        # 执行 Crew
        start_time = time.time()
        result = self.kickoff(inputs=inputs)
        execution_time = time.time() - start_time

        # 收集指标
        metrics = {
            "iteration": i + 1,
            "execution_time": execution_time,
            "total_tokens": result.token_usage.total_tokens,
            "successful_requests": result.token_usage.successful_requests,
            "output_length": len(result.raw),
        }
        results.append(metrics)

        print(f"Execution Time: {execution_time:.2f}s")
        print(f"Total Tokens: {metrics['total_tokens']}")

    # 计算统计数据
    avg_time = sum(r["execution_time"] for r in results) / len(results)
    avg_tokens = sum(r["total_tokens"] for r in results) / len(results)

    print(f"\n=== Test Summary ===")
    print(f"Average Execution Time: {avg_time:.2f}s")
    print(f"Average Total Tokens: {avg_tokens:.0f}")

# 用户代码
result = crew.kickoff(inputs={"topic": "AI Safety"})

# ↓ 进入 Crew.kickoff()
def kickoff(self, inputs: dict[str, Any] | None = None) -> CrewOutput:
    # 1. 设置 OpenTelemetry baggage context
    ctx = baggage.set_baggage("crew_context", CrewContext(id=str(self.id), key=self.key))
    token = attach(ctx)

    try:
        # 2. 执行 before_kickoff 回调
        for before_callback in self.before_kickoff_callbacks:
            inputs = before_callback(inputs)

        # 3. 发送 CrewKickoffStartedEvent
        crewai_event_bus.emit(self, CrewKickoffStartedEvent(crew_name=self.name, inputs=inputs))

        # 4. 重置任务输出处理器
        self._task_output_handler.reset()

        # 5. 插值输入参数
        if inputs is not None:
            self._inputs = inputs
            self._interpolate_inputs(inputs)

        # 6. 设置任务回调
        self._set_tasks_callbacks()

        # 7. 初始化国际化
        i18n = I18N(prompt_file=self.prompt_file)

        # 8. 为每个 Agent 注入资源
        for agent in self.agents:
            agent.i18n = i18n
            agent.crew = self
            agent.set_knowledge(crew_embedder=self.embedder)
            agent.function_calling_llm = self.function_calling_llm
            agent.step_callback = self.step_callback
            agent.create_agent_executor()

        # 9. 处理规划（如果启用）
        if self.planning:
            self._handle_crew_planning()

        # 10. 选择执行流程
        if self.process == Process.sequential:
            result = self._run_sequential_process()
        elif self.process == Process.hierarchical:
            result = self._run_hierarchical_process()

        # 11. 执行 after_kickoff 回调
        for after_callback in self.after_kickoff_callbacks:
            result = after_callback(result)

        # 12. 计算 token 使用
        self.usage_metrics = self.calculate_usage_metrics()

        return result

    except Exception as e:
        crewai_event_bus.emit(self, CrewKickoffFailedEvent(error=str(e), crew_name=self.name))
        raise
    finally:
        detach(token)

# Crew._interpolate_inputs()
def _interpolate_inputs(self, inputs: dict[str, Any]) -> None:
    """将 inputs 插入到任务和 Agent 的模板变量中"""
    # 插值所有任务
    for task in self.tasks:
        task.interpolate_inputs_and_add_conversation_history(inputs)
        # ↓ 进入 Task.interpolate_inputs_and_add_conversation_history()
        #   替换 task.description 中的 {placeholders}
        #   替换 task.expected_output 中的 {placeholders}
        #   处理会话历史（crew_chat_messages）

    # 插值所有 Agent
    for agent in self.agents:
        agent.interpolate_inputs(inputs)
        # ↓ 进入 Agent.interpolate_inputs()
        #   替换 agent.role 中的 {placeholders}
        #   替换 agent.goal 中的 {placeholders}
        #   替换 agent.backstory 中的 {placeholders}

# Crew._run_sequential_process()
def _run_sequential_process(self) -> CrewOutput:
    """顺序执行流程"""
    return self._execute_tasks(self.tasks)
    # ↓ 进入核心任务执行循环

# Crew._execute_tasks()
def _execute_tasks(
    self,
    tasks: list[Task],
    start_index: int | None = 0,
    was_replayed: bool = False,
) -> CrewOutput:
    """任务执行循环"""
    task_outputs: list[TaskOutput] = []
    futures: list[tuple[Task, Future[TaskOutput], int]] = []

    for task_index, task in enumerate(tasks):
        # 跳过已执行的任务（回放场景）
        if start_index is not None and task_index < start_index:
            continue

        # 1. 获取执行 Agent
        agent_to_use = self._get_agent_to_use(task)
        # ↓ Sequential: 返回 task.agent
        # ↓ Hierarchical: 返回 self.manager_agent

        # 2. 准备工具
        tools_for_task = task.tools or agent_to_use.tools or []
        tools_for_task = self._prepare_tools(agent_to_use, task, tools_for_task)
        # ↓ 进入 _prepare_tools()

        # 3. 获取上下文
        context = self._get_context(task, task_outputs)
        # ↓ 进入 _get_context()

        # 4. 处理条件任务
        if isinstance(task, ConditionalTask):
            skipped_output = self._handle_conditional_task(...)
            if skipped_output:
                task_outputs.append(skipped_output)
                continue

        # 5. 执行任务
        if task.async_execution:
            # 异步执行
            future = task.execute_async(agent=agent_to_use, context=context, tools=tools_for_task)
            futures.append((task, future, task_index))
        else:
            # 同步执行
            if futures:
                # 先完成所有异步任务
                task_outputs = self._process_async_tasks(futures)
                futures.clear()

            # 执行当前同步任务
            task_output = task.execute_sync(agent=agent_to_use, context=context, tools=tools_for_task)
            # ↓ 进入 Task.execute_sync()

            task_outputs.append(task_output)
            self._process_task_result(task, task_output)
            self._store_execution_log(task, task_output, task_index, was_replayed)

    # 6. 等待剩余异步任务
    if futures:
        task_outputs = self._process_async_tasks(futures)

    # 7. 创建 CrewOutput
    return self._create_crew_output(task_outputs)

# Crew._prepare_tools()
def _prepare_tools(
    self, agent: BaseAgent, task: Task, tools: list[Tool] | list[BaseTool]
) -> list[BaseTool]:
    """准备工具列表"""
    # 1. 添加委派工具（如果允许）
    if hasattr(agent, "allow_delegation") and getattr(agent, "allow_delegation", False):
        if self.process == Process.hierarchical:
            tools = self._update_manager_tools(task, tools)
            # ↓ 进入 _update_manager_tools()
            #   添加 DelegateTaskTool
        elif agent:
            tools = self._add_delegation_tools(task, tools)
            # ↓ 进入 _add_delegation_tools()
            #   agent.get_delegation_tools(agents)

    # 2. 添加代码执行工具
    if hasattr(agent, "allow_code_execution") and getattr(agent, "allow_code_execution", False):
        tools = self._add_code_execution_tools(agent, tools)
        # ↓ agent.get_code_execution_tools()

    # 3. 添加多模态工具
    if hasattr(agent, "multimodal") and getattr(agent, "multimodal", False):
        tools = self._add_multimodal_tools(agent, tools)
        # ↓ agent.get_multimodal_tools()

    # 4. 添加平台工具
    if hasattr(agent, "apps") and getattr(agent, "apps", None):
        tools = self._add_platform_tools(task, tools)
        # ↓ agent.get_platform_tools(apps)

    # 5. 添加 MCP 工具
    if hasattr(agent, "mcps") and getattr(agent, "mcps", None):
        tools = self._add_mcp_tools(task, tools)
        # ↓ agent.get_mcp_tools(mcps)

    return tools

# Crew._get_context()
def _get_context(self, task: Task, task_outputs: list[TaskOutput]) -> str:
    """获取任务上下文"""
    if not task.context:
        return ""

    if task.context is NOT_SPECIFIED:
        # 默认：聚合所有前序任务的输出
        return aggregate_raw_outputs_from_task_outputs(task_outputs)
        # ↓ 进入 aggregate_raw_outputs_from_task_outputs()
        #   return "\n".join([output.raw for output in task_outputs])
    else:
        # 显式指定：仅聚合特定任务的输出
        return aggregate_raw_outputs_from_tasks(task.context)
        # ↓ 进入 aggregate_raw_outputs_from_tasks()
        #   查找对应的任务输出
        #   return "\n".join([task.output.raw for task in context_tasks if task.output])

# Task.execute_sync()
def execute_sync(
    self,
    agent: BaseAgent | None = None,
    context: str | None = None,
    tools: list[Any] | None = None,
) -> TaskOutput:
    """同步执行任务"""
    return self._execute_core(agent, context, tools)
    # ↓ 进入 _execute_core()

# Task._execute_core()
def _execute_core(
    self,
    agent: BaseAgent | None,
    context: str | None,
    tools: list[Any] | None,
) -> TaskOutput:
    """任务执行核心逻辑"""
    agent = agent or self.agent
    self.agent = agent

    # 1. 记录开始时间
    self.start_time = datetime.datetime.now()

    # 2. 设置上下文
    self.prompt_context = context
    tools = tools or self.tools or []

    # 3. 记录处理 Agent
    self.processed_by_agents.add(agent.role)

    # 4. 发送 TaskStartedEvent
    crewai_event_bus.emit(self, TaskStartedEvent(context=context, task=self))

    # 5. 执行 Agent 任务
    result = agent.execute_task(task=self, context=context, tools=tools)
    # ↓ 进入 Agent.execute_task()

    # 6. 导出输出
    pydantic_output, json_output = self._export_output(result)
    # ↓ 进入 _export_output()
    #   解析 JSON 或 Pydantic 模型

    # 7. 创建 TaskOutput
    task_output = TaskOutput(
        name=self.name or self.description,
        description=self.description,
        expected_output=self.expected_output,
        raw=result,
        pydantic=pydantic_output,
        json_dict=json_output,
        agent=agent.role,
        output_format=self._get_output_format(),
    )

    # 8. 执行 Guardrails 验证
    if self._guardrails:
        for idx, guardrail in enumerate(self._guardrails):
            task_output = self._invoke_guardrail_function(
                task_output=task_output,
                agent=agent,
                tools=tools,
                guardrail=guardrail,
                guardrail_index=idx,
            )
            # ↓ 如果验证失败，重新执行任务

    # 9. 调用回调
    if self.callback:
        self.callback(task_output)

    # 10. 发送 TaskCompletedEvent
    crewai_event_bus.emit(self, TaskCompletedEvent(task_output=task_output, task=self))

    return task_output