stateDiagram-v2
    [*] --> 创建: Task(...) 或 @task 装饰器
    创建 --> 配置: 设置 description/expected_output
    配置 --> 插值: interpolate_inputs(inputs)
    插值 --> 就绪: 等待执行
    就绪 --> 执行中: execute_sync() 或 execute_async()
    执行中 --> Agent执行: agent.execute_task()
    Agent执行 --> 输出转换: _export_output()
    输出转换 --> Guardrail验证: 如果配置了 guardrail
    Guardrail验证 --> Guardrail验证: 验证失败，重新执行
    Guardrail验证 --> 输出保存: 验证通过
    输出保存 --> 文件持久化: 如果配置了 output_file
    文件持久化 --> 完成: TaskOutput
    完成 --> [*]

flowchart TB
    subgraph "上游编排层"
        Crew[Crew 编排器<br/>任务序列管理]
        Flow[Flow 引擎<br/>流程编排]
    end

    subgraph "Task 核心层"
        Task[Task 类<br/>任务定义和配置]
        ConditionalTask[ConditionalTask 类<br/>条件执行任务]
        TaskAPI{Task API}
        ExecuteSync[execute_sync<br/>同步执行]
        ExecuteAsync[execute_async<br/>异步执行]
        Prompt[prompt<br/>提示词生成]
        Interpolate[interpolate_inputs<br/>输入插值]
    end

    subgraph "执行引擎层"
        Agent[Agent 代理<br/>任务执行主体]
        AgentExecutor[AgentExecutor<br/>执行器]
        LLM[LLM 语言模型<br/>推理引擎]
        Tools[Tools 工具<br/>外部能力]
    end

    subgraph "输出处理层"
        TaskOutput[TaskOutput<br/>输出封装]
        Converter[Converter<br/>结构化转换]
        OutputFormat[OutputFormat<br/>格式枚举]
        FileHandler[FileHandler<br/>文件持久化]
    end

    subgraph "验证层"
        Guardrail[Guardrail<br/>输出验证]
        GuardrailResult[GuardrailResult<br/>验证结果]
    end

    subgraph "基础设施层"
        Events[EventBus<br/>事件总线]
        Memory[Memory<br/>记忆系统]
        RPM[RPMController<br/>速率限制]
        Security[SecurityConfig<br/>安全配置]
    end

    Crew -->|kickoff| Task
    Flow -->|调用| Task

    Task --> TaskAPI
    TaskAPI --> ExecuteSync
    TaskAPI --> ExecuteAsync
    TaskAPI --> Prompt
    TaskAPI --> Interpolate

    ExecuteSync --> Agent
    ExecuteAsync --> Agent

    Agent --> AgentExecutor
    AgentExecutor --> LLM
    AgentExecutor --> Tools

    Agent -->|返回结果| Task
    Task --> Converter
    Converter --> TaskOutput
    TaskOutput --> OutputFormat

    Task --> Guardrail
    Guardrail -->|验证失败| Agent
    Guardrail --> GuardrailResult

    TaskOutput --> FileHandler

    Task --> Events
    Agent --> Events
    Agent --> Memory
    Agent --> RPM
    Task --> Security

    ConditionalTask -.继承.-> Task

flowchart TB
    subgraph "Task 核心层"
        Task[Task 类<br/>任务定义和配置]
        ConditionalTask[ConditionalTask 类<br/>条件执行任务]
    end

    subgraph "输出处理层"
        TaskOutput[TaskOutput 类<br/>任务输出封装]
        Converter[Converter 类<br/>结构化输出转换]
        OutputFormat[OutputFormat 枚举<br/>RAW/JSON/PYDANTIC]
    end

    subgraph "验证层"
        Guardrail[Guardrail 机制<br/>输出验证]
        LLMGuardrail[LLMGuardrail 类<br/>基于 LLM 的验证]
        HallucinationGuardrail[HallucinationGuardrail<br/>幻觉检测]
    end

    subgraph "执行层"
        ExecuteSync[execute_sync()<br/>同步执行]
        ExecuteAsync[execute_async()<br/>异步执行]
        Agent[Agent.execute_task()]
    end

    subgraph "上下文层"
        Context[Context 管理<br/>前序任务输出]
        Interpolation[输入插值<br/>模板变量替换]
    end

    subgraph "基础设施层"
        Events[事件系统<br/>TaskStarted/Completed/Failed]
        Security[SecurityConfig<br/>指纹验证]
        FileHandler[文件处理<br/>输出持久化]
    end

    Task --> ExecuteSync
    Task --> ExecuteAsync
    ConditionalTask -.继承.-> Task

    ExecuteSync --> Agent
    ExecuteAsync --> Agent

    Agent --> TaskOutput
    TaskOutput --> Converter
    TaskOutput --> OutputFormat

    Task --> Guardrail
    Guardrail --> LLMGuardrail
    Guardrail --> HallucinationGuardrail

    Task --> Context
    Task --> Interpolation

    Task --> Events
    Task --> Security
    Task --> FileHandler

用户输入 → Crew.kickoff → Task.execute_sync → Agent.execute_task → LLM.call → 原始输出
→ Converter.convert → TaskOutput → Guardrail.validate → FileHandler.save → CrewOutput

TaskStartedEvent → AgentExecutionStartedEvent → LLMGuardrailStartedEvent
→ LLMGuardrailCompletedEvent → AgentExecutionCompletedEvent → TaskCompletedEvent

flowchart TD
    Start([用户调用 Crew.kickoff]) --> KickoffInput[传入 inputs 参数]
    KickoffInput --> InterpolateInputs[Crew: 对所有任务插值<br/>interpolate_inputs_and_add_conversation_history]
    InterpolateInputs --> ExecuteTasks[Crew: _execute_tasks<br/>遍历任务列表]

    ExecuteTasks --> LoopStart{遍历每个任务}

    LoopStart --> GetAgent[Crew: _get_agent_to_use<br/>确定执行该任务的 Agent]
    GetAgent --> PrepareTools[Crew: _prepare_tools<br/>准备工具列表]
    PrepareTools --> GetContext[Crew: _get_context<br/>获取前序任务输出作为上下文]

    GetContext --> CheckConditional{是否是<br/>ConditionalTask?}
    CheckConditional -->|是| CheckCondition[Crew: _handle_conditional_task<br/>检查执行条件]
    CheckCondition --> ShouldExecute{条件满足?}
    ShouldExecute -->|否| SkipTask[跳过任务<br/>返回空输出]
    ShouldExecute -->|是| CheckAsync

    CheckConditional -->|否| CheckAsync{async_execution?}

    CheckAsync -->|是| ExecuteAsync[Task: execute_async<br/>异步执行]
    ExecuteAsync --> AddToFutures[将 Future 加入队列]
    AddToFutures --> LoopStart

    CheckAsync -->|否| WaitForAsyncTasks{有待处理的<br/>异步任务?}
    WaitForAsyncTasks -->|是| ProcessAsyncTasks[Crew: _process_async_tasks<br/>等待所有异步任务完成]
    ProcessAsyncTasks --> ExecuteSync
    WaitForAsyncTasks -->|否| ExecuteSync[Task: execute_sync<br/>同步执行]

    ExecuteSync --> TaskCore[Task: _execute_core<br/>核心执行逻辑]
    TaskCore --> AgentExecute[Agent: execute_task<br/>Agent 执行任务]
    AgentExecute --> ExportOutput[Task: _export_output<br/>转换输出格式]
    ExportOutput --> CreateTaskOutput[创建 TaskOutput 对象]

    CreateTaskOutput --> HasGuardrail{配置了<br/>Guardrail?}
    HasGuardrail -->|是| InvokeGuardrail[Task: _invoke_guardrail_function<br/>执行验证]
    InvokeGuardrail --> GuardrailPass{验证通过?}
    GuardrailPass -->|否| RetryTask[重新执行任务<br/>最多重试 N 次]
    RetryTask --> AgentExecute
    GuardrailPass -->|是| SaveOutput
    HasGuardrail -->|否| SaveOutput

    SaveOutput[保存任务输出] --> HasOutputFile{配置了<br/>output_file?}
    HasOutputFile -->|是| SaveFile[Task: _save_file<br/>持久化到文件]
    HasOutputFile -->|否| ExecuteCallback
    SaveFile --> ExecuteCallback[执行回调函数]

    ExecuteCallback --> EmitCompleted[发送 TaskCompletedEvent]
    EmitCompleted --> AddToOutputs[将输出加入 task_outputs 列表]
    AddToOutputs --> LoopStart

    SkipTask --> AddToOutputs

    LoopStart -->|所有任务完成| FinalAsyncCheck{还有未完成的<br/>异步任务?}
    FinalAsyncCheck -->|是| FinalProcessAsync[Crew: _process_async_tasks<br/>等待最后的异步任务]
    FinalProcessAsync --> CreateCrewOutput
    FinalAsyncCheck -->|否| CreateCrewOutput[Crew: _create_crew_output<br/>创建 CrewOutput]

    CreateCrewOutput --> End([返回 CrewOutput])

flowchart TD
    Start([Task.execute_sync]) --> CheckAgent{Agent<br/>已分配?}
    CheckAgent -->|否| ThrowError[抛出异常:<br/>No agent assigned]
    CheckAgent -->|是| RecordStartTime[记录 start_time]

    RecordStartTime --> SetContext[设置 prompt_context]
    SetContext --> AddToProcessed[将 Agent 角色加入<br/>processed_by_agents]
    AddToProcessed --> EmitStarted[发送 TaskStartedEvent]

    EmitStarted --> CallAgent[调用 Agent.execute_task<br/>传入 task/context/tools]
    CallAgent --> AgentInternalFlow[Agent 内部执行流程<br/>见下一节]
    AgentInternalFlow --> GetResult[获取 Agent 返回的<br/>字符串结果]

    GetResult --> ExportOutput[Task: _export_output<br/>转换输出格式]

    ExportOutput --> HasPydantic{配置了<br/>output_pydantic?}
    HasPydantic -->|是| ConvertToPydantic[Converter: to_pydantic<br/>转换为 Pydantic 模型]
    ConvertToPydantic --> CreateOutput

    HasPydantic -->|否| HasJSON{配置了<br/>output_json?}
    HasJSON -->|是| ConvertToJSON[Converter: to_json<br/>转换为 JSON 字典]
    ConvertToJSON --> CreateOutput
    HasJSON -->|否| CreateOutput[创建 TaskOutput 对象]

    CreateOutput --> HasGuardrails{配置了<br/>guardrails 列表?}
    HasGuardrails -->|是| LoopGuardrails[遍历每个 Guardrail]
    LoopGuardrails --> InvokeGuardrail[Task: _invoke_guardrail_function<br/>执行单个 Guardrail]

    InvokeGuardrail --> ProcessGuardrail[process_guardrail<br/>调用验证函数]
    ProcessGuardrail --> EmitGuardrailStarted[发送 LLMGuardrailStartedEvent]
    EmitGuardrailStarted --> CallGuardrailFunc[调用用户定义的<br/>guardrail task_output]
    CallGuardrailFunc --> EmitGuardrailCompleted[发送 LLMGuardrailCompletedEvent]

    EmitGuardrailCompleted --> GuardrailSuccess{验证成功?}
    GuardrailSuccess -->|是| UpdateOutput[更新 task_output]
    UpdateOutput --> NextGuardrail{还有其他<br/>Guardrail?}
    NextGuardrail -->|是| LoopGuardrails
    NextGuardrail -->|否| CheckSingleGuardrail

    GuardrailSuccess -->|否| CheckRetry{重试次数<br/>< max_retries?}
    CheckRetry -->|是| IncrementRetry[增加 retry_count]
    IncrementRetry --> PrintRetry[打印重试信息]
    PrintRetry --> BuildErrorContext[构建错误上下文]
    BuildErrorContext --> CallAgentAgain[重新调用 Agent.execute_task]
    CallAgentAgain --> AgentInternalFlow

    CheckRetry -->|否| ThrowGuardrailError[抛出异常:<br/>Guardrail validation failed]

    HasGuardrails -->|否| CheckSingleGuardrail{配置了<br/>单个 guardrail?}
    CheckSingleGuardrail -->|是| InvokeGuardrail
    CheckSingleGuardrail -->|否| SaveToSelf[self.output = task_output]

    SaveToSelf --> RecordEndTime[记录 end_time]
    RecordEndTime --> ExecuteCallback{配置了<br/>callback?}
    ExecuteCallback -->|是| CallCallback[执行 self.callback]
    CallCallback --> CheckCrewCallback
    ExecuteCallback -->|否| CheckCrewCallback{Crew 配置了<br/>task_callback?}

    CheckCrewCallback -->|是| CallCrewCallback[执行 crew.task_callback]
    CheckCrewCallback -->|否| CheckOutputFile
    CallCrewCallback --> CheckOutputFile{配置了<br/>output_file?}

    CheckOutputFile -->|是| PrepareContent[准备文件内容:<br/>JSON/Pydantic/Raw]
    PrepareContent --> SaveFile[Task: _save_file<br/>保存到文件系统]
    CheckOutputFile -->|否| EmitCompleted
    SaveFile --> EmitCompleted[发送 TaskCompletedEvent]

    EmitCompleted --> ReturnOutput[返回 TaskOutput]
    ReturnOutput --> End([结束])

    ThrowError --> End
    ThrowGuardrailError --> EmitFailed[发送 TaskFailedEvent]
    EmitFailed --> End

flowchart TD
    Start([Agent.execute_task]) --> HasReasoning{启用了<br/>Reasoning?}
    HasReasoning -->|是| CreateReasoningHandler[创建 AgentReasoning<br/>推理处理器]
    CreateReasoningHandler --> GeneratePlan[生成推理计划<br/>handle_agent_reasoning]
    GeneratePlan --> AddPlanToTask[将计划添加到<br/>task.description]
    AddPlanToTask --> InjectDate

    HasReasoning -->|否| InjectDate[注入当前日期到任务]

    InjectDate --> ResetToolsHandler[重置 tools_handler.last_used_tool]
    ResetToolsHandler --> GetTaskPrompt[获取任务提示词<br/>task.prompt]

    GetTaskPrompt --> HasOutputFormat{需要结构化<br/>输出?}
    HasOutputFormat -->|是| AddFormatInstructions[添加输出格式指令<br/>JSON/Pydantic schema]
    AddFormatInstructions --> PrepareExecution
    HasOutputFormat -->|否| PrepareExecution[准备执行环境]

    PrepareExecution --> CreateMessages[构建消息列表:<br/>system + user + history]
    CreateMessages --> SetupCallbacks[设置 TokenCalcHandler<br/>回调]

    SetupCallbacks --> EmitExecutionStarted[发送 AgentExecutionStartedEvent]
    EmitExecutionStarted --> StartIterationLoop[开始推理循环<br/>iteration = 0]

    StartIterationLoop --> CheckMaxIter{iteration >=<br/>max_iter?}
    CheckMaxIter -->|是| HandleMaxIterations[处理最大迭代次数<br/>生成最终答案]
    HandleMaxIterations --> FormatFinalAnswer

    CheckMaxIter -->|否| CheckRPM[RPMController:<br/>检查速率限制]
    CheckRPM --> WaitIfNeeded{需要等待?}
    WaitIfNeeded -->|是| Sleep[等待一段时间]
    Sleep --> CallLLM
    WaitIfNeeded -->|否| CallLLM[调用 LLM.call<br/>发送消息]

    CallLLM --> GetLLMResponse[get_llm_response<br/>获取 LLM 响应]
    GetLLMResponse --> ParseResponse[解析响应:<br/>AgentAction 或 AgentFinish?]

    ParseResponse --> IsAction{是 AgentAction?}
    IsAction -->|是| ExecuteTool[execute_tool_and_check_finality<br/>执行工具调用]

    ExecuteTool --> ToolSuccess{工具执行<br/>成功?}
    ToolSuccess -->|是| AddToolResult[将工具结果添加到<br/>messages]
    AddToolResult --> IsToolFinal{工具返回<br/>final_answer?}
    IsToolFinal -->|是| FormatFinalAnswer
    IsToolFinal -->|否| IncrementIter[iteration++]
    IncrementIter --> StartIterationLoop

    ToolSuccess -->|否| HandleToolError[处理工具错误]
    HandleToolError --> AddErrorMessage[将错误信息添加到<br/>messages]
    AddErrorMessage --> IncrementIter

    IsAction -->|否| IsFinish{是 AgentFinish?}
    IsFinish -->|是| FormatFinalAnswer[格式化最终答案]

    IsFinish -->|否| HandleParseError[处理解析错误<br/>OutputParserError]
    HandleParseError --> AddParseErrorMsg[添加解析错误提示]
    AddParseErrorMsg --> IncrementIter

    FormatFinalAnswer --> AddToMemory{启用了<br/>Memory?}
    AddToMemory -->|是| SaveToMemory[保存到记忆系统:<br/>短期/长期/实体]
    SaveToMemory --> EmitExecutionCompleted

    AddToMemory -->|否| EmitExecutionCompleted[发送 AgentExecutionCompletedEvent]
    EmitExecutionCompleted --> ReturnResult[返回字符串结果]
    ReturnResult --> End([结束])

# 文件：lib/crewai/src/crewai/crew.py
def kickoff(self, inputs: dict = None) -> CrewOutput:
    # 1. 插值所有任务的输入变量
    for task in self.tasks:
        task.interpolate_inputs_and_add_conversation_history(inputs)

    # 2. 执行任务序列
    return self._execute_tasks(self.tasks)

# 文件：lib/crewai/src/crewai/crew.py
def _execute_tasks(self, tasks: list[Task], start_index: int = 0) -> CrewOutput:
    task_outputs: list[TaskOutput] = []
    futures: list[(Task, Future, int)] = []

    for task_index, task in enumerate(tasks):
        # 1. 确定执行该任务的 Agent
        agent_to_use = self._get_agent_to_use(task)

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

        # 3. 获取上下文（前序任务的输出）
        context = self._get_context(task, task_outputs)

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

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

            # 执行同步任务
            task_output = task.execute_sync(agent_to_use, context, tools_for_task)
            task_outputs.append(task_output)

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

    return self._create_crew_output(task_outputs)

# 文件：lib/crewai/src/crewai/task.py
def execute_sync(self, agent: BaseAgent, context: str, tools: list[BaseTool]) -> TaskOutput:
    return self._execute_core(agent, context, tools)

def _execute_core(self, agent: BaseAgent, context: str, tools: list[Any]) -> TaskOutput:
    # 1. 验证 Agent
    if not agent:
        raise Exception(f"The task '{self.description}' has no agent assigned")

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

    # 3. 发送任务开始事件
    crewai_event_bus.emit(self, TaskStartedEvent(context=context, task=self))

    # 4. 调用 Agent 执行任务
    result = agent.execute_task(task=self, context=context, tools=tools)

    # 5. 转换输出格式
    pydantic_output, json_output = self._export_output(result)

    # 6. 创建 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(),
    )

    # 7. 执行 Guardrail 验证
    if self._guardrails:
        for idx, guardrail in enumerate(self._guardrails):
            task_output = self._invoke_guardrail_function(
                task_output, agent, tools, guardrail, idx
            )

    # 8. 保存输出
    self.output = task_output
    self.end_time = datetime.datetime.now()

    # 9. 执行回调
    if self.callback:
        self.callback(self.output)

    # 10. 保存到文件
    if self.output_file:
        self._save_file(content)

    # 11. 发送任务完成事件
    crewai_event_bus.emit(self, TaskCompletedEvent(output=task_output, task=self))

    return task_output

# 文件：lib/crewai/src/crewai/agent.py
def execute_task(self, task: Task, context: str, tools: list[BaseTool]) -> str:
    # 1. 处理推理计划（如果启用）
    if self.reasoning:
        reasoning_handler = AgentReasoning(task=task, agent=self)
        reasoning_output = reasoning_handler.handle_agent_reasoning()
        task.description += f"\n\nReasoning Plan:\n{reasoning_output.plan.plan}"

    # 2. 注入当前日期
    self._inject_date_to_task(task)

    # 3. 生成任务提示词
    task_prompt = task.prompt()

    # 4. 添加上下文
    if context:
        task_prompt = self.i18n.slice("task_with_context").format(
            task=task_prompt, context=context
        )

    # 5. 添加输出格式指令
    if task.output_pydantic or task.output_json:
        model = task.output_pydantic or task.output_json
        task_prompt += f"\n\nOutput schema:\n{model.model_json_schema()}"

    # 6. 进入推理循环
    return self._run_agent_executor(task_prompt, tools)

def _run_agent_executor(self, task_prompt: str, tools: list[BaseTool]) -> str:
    messages = self._build_messages(task_prompt)
    callbacks = [TokenCalcHandler()]
    iteration = 0

    while iteration < self.max_iter:
        # 检查速率限制
        self._enforce_rpm_limit()

        # 调用 LLM
        answer = self.llm.call(messages, callbacks=callbacks)

        # 解析响应
        formatted_answer = self._parse_llm_response(answer)

        # 处理 AgentAction（工具调用）
        if isinstance(formatted_answer, AgentAction):
            tool_result = self._execute_tool(formatted_answer, tools)
            messages.append({"role": "assistant", "content": answer})
            messages.append({"role": "user", "content": tool_result})

            # 检查工具是否返回最终答案
            if self._is_final_answer(tool_result):
                return tool_result

            iteration += 1
            continue

        # 处理 AgentFinish（任务完成）
        if isinstance(formatted_answer, AgentFinish):
            return formatted_answer.output

        iteration += 1

    # 达到最大迭代次数
    return self._handle_max_iterations_exceeded(messages)

# 文件：lib/crewai/src/crewai/llm.py
def call(self, messages: list[dict], callbacks: list = None) -> str:
    # 1. 准备调用参数
    params = {
        "model": self.model,
        "messages": messages,
        "temperature": self.temperature,
        "max_tokens": self.max_tokens,
    }

    # 2. 添加工具定义（如果支持 function calling）
    if self.supports_function_calling():
        params["tools"] = self._format_tools()

    # 3. 调用 LLM API
    response = self.client.chat.completions.create(**params)

    # 4. 解析响应
    if response.choices[0].message.tool_calls:
        # 返回工具调用
        tool_call = response.choices[0].message.tool_calls[0]
        return f"Action: {tool_call.function.name}\nAction Input: {tool_call.function.arguments}"
    else:
        # 返回文本回复
        return response.choices[0].message.content

# 文件：lib/crewai/src/crewai/task.py
def execute_async(self, agent: BaseAgent, context: str, tools: list[BaseTool]) -> Future[TaskOutput]:
    future: Future[TaskOutput] = Future()

    # 在新线程中执行
    threading.Thread(
        daemon=True,
        target=self._execute_task_async,
        args=(agent, context, tools, future),
    ).start()

    return future

def _execute_task_async(self, agent, context, tools, future):
    # 调用核心执行逻辑
    result = self._execute_core(agent, context, tools)
    # 设置 Future 结果
    future.set_result(result)

def _process_async_tasks(self, futures: list[(Task, Future, int)]) -> list[TaskOutput]:
    task_outputs = []

    for task, future, task_index in futures:
        # 阻塞等待 Future 完成
        task_output = future.result()
        task_outputs.append(task_output)

        # 处理任务结果
        self._process_task_result(task, task_output)
        self._store_execution_log(task, task_output, task_index)

    return task_outputs

# 文件：lib/crewai/src/crewai/task.py
def _invoke_guardrail_function(
    self,
    task_output: TaskOutput,
    agent: BaseAgent,
    tools: list[BaseTool],
    guardrail: GuardrailCallable,
    guardrail_index: int = None,
) -> TaskOutput:
    current_retry_count = self._guardrail_retry_counts.get(guardrail_index, 0)
    max_attempts = self.guardrail_max_retries + 1

    for attempt in range(max_attempts):
        # 执行验证
        guardrail_result = process_guardrail(
            output=task_output,
            guardrail=guardrail,
            retry_count=current_retry_count,
            event_source=self,
            from_task=self,
            from_agent=agent,
        )

        # 验证通过
        if guardrail_result.success:
            # 更新输出
            if isinstance(guardrail_result.result, str):
                task_output.raw = guardrail_result.result
                pydantic_output, json_output = self._export_output(guardrail_result.result)
                task_output.pydantic = pydantic_output
                task_output.json_dict = json_output
            elif isinstance(guardrail_result.result, TaskOutput):
                task_output = guardrail_result.result

            return task_output

        # 验证失败
        if attempt >= self.guardrail_max_retries:
            raise Exception(
                f"Task failed guardrail validation after {self.guardrail_max_retries} retries. "
                f"Last error: {guardrail_result.error}"
            )

        # 更新重试次数
        current_retry_count += 1
        self._guardrail_retry_counts[guardrail_index] = current_retry_count

        # 构建错误上下文
        context = self.i18n.errors("validation_error").format(
            guardrail_result_error=guardrail_result.error,
            task_output=task_output.raw,
        )

        # 打印重试信息
        printer.print(
            f"Guardrail blocked (attempt {attempt + 1}/{max_attempts}), "
            f"retrying due to: {guardrail_result.error}\n",
            color="yellow",
        )

        # 重新执行任务
        result = agent.execute_task(task=self, context=context, tools=tools)

        # 重新转换输出
        pydantic_output, json_output = self._export_output(result)
        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(),
        )

    return task_output

# 文件：lib/crewai/src/crewai/utilities/guardrail.py
def process_guardrail(
    output: TaskOutput,
    guardrail: GuardrailCallable,
    retry_count: int,
    event_source: Any,
    from_agent: BaseAgent,
    from_task: Task,
) -> GuardrailResult:
    # 发送开始事件
    crewai_event_bus.emit(
        event_source,
        LLMGuardrailStartedEvent(
            guardrail=guardrail,
            retry_count=retry_count,
            from_agent=from_agent,
            from_task=from_task,
        ),
    )

    # 调用验证函数
    result = guardrail(output)  # 返回 (bool, Any)
    guardrail_result = GuardrailResult.from_tuple(result)

    # 发送完成事件
    crewai_event_bus.emit(
        event_source,
        LLMGuardrailCompletedEvent(
            success=guardrail_result.success,
            result=guardrail_result.result,
            error=guardrail_result.error,
            retry_count=retry_count,
            from_agent=from_agent,
            from_task=from_task,
        ),
    )

    return guardrail_result

# 文件：lib/crewai/src/crewai/task.py
def _export_output(self, result: str) -> tuple[BaseModel | None, dict | None]:
    pydantic_output = None
    json_output = None

    if self.output_pydantic or self.output_json:
        model_output = convert_to_model(
            result,
            self.output_pydantic,
            self.output_json,
            self.agent,
            self.converter_cls,
        )

        if isinstance(model_output, BaseModel):
            pydantic_output = model_output
        elif isinstance(model_output, dict):
            json_output = model_output
        elif isinstance(model_output, str):
            try:
                json_output = json.loads(model_output)
            except json.JSONDecodeError:
                json_output = None

    return pydantic_output, json_output

# 文件：lib/crewai/src/crewai/utilities/converter.py
def convert_to_model(
    result: str,
    output_pydantic: type[BaseModel],
    output_json: type[BaseModel],
    agent: Agent,
    converter_cls: type[Converter] = None,
) -> BaseModel | dict | str:
    model = output_pydantic or output_json

    # 尝试直接解析 JSON
    try:
        escaped_result = json.dumps(json.loads(result, strict=False))
        return validate_model(escaped_result, model, bool(output_json))
    except json.JSONDecodeError:
        # JSON 解析失败，使用 Converter
        return handle_partial_json(result, model, bool(output_json), agent, converter_cls)

def handle_partial_json(
    result: str,
    model: type[BaseModel],
    is_json_output: bool,
    agent: Agent,
    converter_cls: type[Converter] = None,
) -> BaseModel | dict | str:
    # 尝试提取部分 JSON
    try:
        json_text = extract_json_from_text(result)
        return validate_model(json_text, model, is_json_output)
    except Exception:
        # 提取失败，使用 LLM 转换
        return convert_with_instructions(result, model, is_json_output, agent, converter_cls)

def convert_with_instructions(
    result: str,
    model: type[BaseModel],
    is_json_output: bool,
    agent: Agent,
    converter_cls: type[Converter] = None,
) -> BaseModel | dict | str:
    llm = agent.function_calling_llm or agent.llm
    instructions = get_conversion_instructions(model, llm)

    converter = Converter(llm=llm, text=result, model=model, instructions=instructions)

    if is_json_output:
        return converter.to_json()
    else:
        return converter.to_pydantic()

class Converter:
    def to_pydantic(self, current_attempt: int = 1) -> BaseModel:
        if self.llm.supports_function_calling():
            # 使用 function calling 直接获取结构化输出
            return self._create_instructor().to_pydantic()
        else:
            # 使用提示词让 LLM 输出 JSON
            response = self.llm.call([
                {"role": "system", "content": self.instructions},
                {"role": "user", "content": self.text},
            ])

            # 验证并解析
            return self.model.model_validate_json(response)

classDiagram
    class Task {
        +UUID id
        +str name
        +str description
        +str expected_output
        +BaseAgent agent
        +list[Task] | _NotSpecified context
        +bool async_execution
        +type[BaseModel] output_json
        +type[BaseModel] output_pydantic
        +str output_file
        +bool create_directory
        +TaskOutput output
        +list[BaseTool] tools
        +SecurityConfig security_config
        +bool human_input
        +bool markdown
        +type[Converter] converter_cls
        +set[str] processed_by_agents
        +GuardrailType guardrail
        +GuardrailsType guardrails
        +int guardrail_max_retries
        +int retry_count
        +datetime start_time
        +datetime end_time
        +bool allow_crewai_trigger_context
        +execute_sync(agent, context, tools)
        +execute_async(agent, context, tools)
        +prompt()
        +interpolate_inputs_and_add_conversation_history(inputs)
        +copy(agents, task_mapping)
        -_execute_core(agent, context, tools)
        -_export_output(result)
        -_save_file(result)
        -_invoke_guardrail_function(task_output, agent, tools, guardrail)
    }

    class ConditionalTask {
        +Callable[[TaskOutput], bool] condition
        +should_execute(context)
        +get_skipped_task_output()
    }

    class TaskOutput {
        +str description
        +str name
        +str expected_output
        +str summary
        +str raw
        +BaseModel pydantic
        +dict json_dict
        +str agent
        +OutputFormat output_format
        +json()
        +to_dict()
    }

    class OutputFormat {
        <<enumeration>>
        RAW
        JSON
        PYDANTIC
    }

    class Converter {
        +BaseLLM llm
        +str text
        +type[BaseModel] model
        +str instructions
        +convert()
    }

    Task <|-- ConditionalTask : 继承
    Task --> TaskOutput : 生成
    TaskOutput --> OutputFormat : 使用
    Task --> Converter : 使用

class ExecuteSyncParams:
    agent: Optional[BaseAgent] = None     # 执行任务的 Agent（可覆盖默认）
    context: Optional[str] = None         # 上下文字符串
    tools: Optional[list[BaseTool]] = None  # 工具列表

class TaskOutput(BaseModel):
    description: str                # 任务描述
    name: Optional[str]             # 任务名称
    expected_output: Optional[str]  # 预期输出
    summary: Optional[str]          # 任务摘要
    raw: str                        # 原始输出
    pydantic: Optional[BaseModel]   # Pydantic 模型输出
    json_dict: Optional[dict]       # JSON 字典输出
    agent: str                      # 执行的 Agent 角色
    output_format: OutputFormat     # 输出格式

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

def _execute_core(
    self,
    agent: BaseAgent | None,
    context: str | None,
    tools: list[Any] | None,
) -> TaskOutput:
    """核心执行逻辑"""
    try:
        # 1. 确定执行 Agent
        agent = agent or self.agent
        self.agent = agent
        if not agent:
            raise Exception(
                f"The task '{self.description}' has no agent assigned"
            )

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

        # 3. 设置提示词上下文
        self.prompt_context = context
        tools = tools or self.tools or []

        # 4. 记录处理该任务的 Agent
        self.processed_by_agents.add(agent.role)

        # 5. 发送任务开始事件
        crewai_event_bus.emit(self, TaskStartedEvent(context=context, task=self))

        # 6. 调用 Agent 执行任务
        result = agent.execute_task(
            task=self,
            context=context,
            tools=tools,
        )

        # 7. 导出结构化输出
        pydantic_output, json_output = self._export_output(result)

        # 8. 创建 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(),
        )

        # 9. 执行 Guardrail 验证（如果配置）
        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,
                )

        # 10. 执行单个 Guardrail（向后兼容）
        if self._guardrail:
            task_output = self._invoke_guardrail_function(
                task_output=task_output,
                agent=agent,
                tools=tools,
                guardrail=self._guardrail,
            )

        # 11. 保存输出
        self.output = task_output
        self.end_time = datetime.datetime.now()

        # 12. 执行回调函数
        if self.callback:
            self.callback(self.output)

        crew = self.agent.crew
        if crew and crew.task_callback and crew.task_callback != self.callback:
            crew.task_callback(self.output)

        # 13. 保存到文件（如果配置）
        if self.output_file:
            content = (
                json_output
                if json_output
                else (
                    pydantic_output.model_dump_json() if pydantic_output else result
                )
            )
            self._save_file(content)

        # 14. 发送任务完成事件
        crewai_event_bus.emit(
            self, TaskCompletedEvent(output=task_output, task=self)
        )

        return task_output

    except Exception as e:
        self.end_time = datetime.datetime.now()
        crewai_event_bus.emit(self, TaskFailedEvent(error=str(e), task=self))
        raise e

# 上游调用者：Crew._execute_tasks()
def _execute_tasks(self, tasks: list[Task], start_index: int = 0) -> CrewOutput:
    """Crew 执行任务序列"""
    task_outputs: list[TaskOutput] = []

    for task_index, task in enumerate(tasks):
        agent_to_use = self._get_agent_to_use(task)
        tools_for_task = self._prepare_tools(agent_to_use, task, task.tools or [])

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

        # 执行任务
        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:
            # 同步执行
            task_output = task.execute_sync(
                agent=agent_to_use,
                context=context,
                tools=tools_for_task,
            )
            task_outputs.append(task_output)
            self._process_task_result(task, task_output)

    return self._create_crew_output(task_outputs)

sequenceDiagram
    autonumber
    participant Crew as Crew
    participant Task as Task
    participant Agent as Agent
    participant Converter as Converter
    participant Guardrail as Guardrail
    participant FileHandler as FileHandler

    Crew->>Task: execute_sync(agent, context, tools)
    Task->>Task: 记录 start_time
    Task->>Task: 发送 TaskStartedEvent

    Task->>Agent: execute_task(task, context, tools)
    Agent->>Agent: 推理循环
    Agent-->>Task: 返回 result（字符串）

    Task->>Converter: _export_output(result)
    alt 配置了 output_pydantic
        Converter->>Converter: convert_to_model(result, output_pydantic)
        Converter-->>Task: pydantic_output
    else 配置了 output_json
        Converter->>Converter: convert_to_model(result, output_json)
        Converter-->>Task: json_output
    else 默认
        Converter-->>Task: None
    end

    Task->>Task: 创建 TaskOutput

    alt 配置了 guardrail
        Task->>Guardrail: _invoke_guardrail_function(task_output)
        Guardrail->>Guardrail: process_guardrail(output)
        alt 验证失败
            Guardrail->>Agent: 重新执行任务
            Agent-->>Guardrail: 新的 result
            Guardrail->>Converter: 重新转换
        end
        Guardrail-->>Task: 验证通过的 task_output
    end

    Task->>Task: 执行回调函数

    alt 配置了 output_file
        Task->>FileHandler: _save_file(content)
        FileHandler-->>Task: 保存成功
    end

    Task->>Task: 记录 end_time
    Task->>Task: 发送 TaskCompletedEvent
    Task-->>Crew: TaskOutput

@property
def execution_duration(self) -> float | None:
    """计算任务执行时长（秒）"""
    if not self.start_time or not self.end_time:
        return None
    return (self.end_time - self.start_time).total_seconds()

def execute_async(
    self,
    agent: BaseAgent | None = None,
    context: str | None = None,
    tools: list[BaseTool] | None = None,
) -> Future[TaskOutput]:
    """异步执行任务"""
    future: Future[TaskOutput] = Future()
    threading.Thread(
        daemon=True,
        target=self._execute_task_async,
        args=(agent, context, tools, future),
    ).start()
    return future

def _execute_task_async(
    self,
    agent: BaseAgent | None,
    context: str | None,
    tools: list[Any] | None,
    future: Future[TaskOutput],
) -> None:
    """异步执行的内部方法"""
    result = self._execute_core(agent, context, tools)
    future.set_result(result)