第一步，把 CrewAI 工具翻译成 OpenAI 格式（utilities/agent_utils.py:154）：

# utilities/agent_utils.py:154
def convert_tools_to_openai_schema(tools):
    """Convert CrewAI tools to OpenAI function calling format. ...
    Returns: (schema 列表, 名字→可调用, 名字→原始工具对象)"""
    openai_tools, available_functions, tool_name_mapping = [], {}, {}
    for tool in tools:
        parameters = {}
        if hasattr(tool, "args_schema") and tool.args_schema is not None:
            schema_output = generate_model_description(tool.args_schema, strip_null_types=False)
            parameters = schema_output.get("json_schema", {}).get("schema", {})
            parameters.pop("title", None); parameters.pop("description", None)
        description = tool.description
        if "Tool Description:" in description:
            description = description.split("Tool Description:")[-1].strip()  # 抠出纯描述
        sanitized_name = sanitize_tool_name(tool.name)          # 清洗成合法函数名
        if sanitized_name in available_functions:               # ★同名冲突→加后缀
            counter = 2
            candidate = sanitize_tool_name(f"{sanitized_name}_{counter}")
            while candidate in available_functions:
                counter += 1; candidate = sanitize_tool_name(f"{sanitized_name}_{counter}")
            sanitized_name = candidate
        schema = {"type": "function", "function": {
            "name": sanitized_name, "description": description,
            "parameters": parameters, "strict": True}}          # strict=严格模式

转好的 schema，通过 _prepare_completion_params 塞进请求（llm.py:748）：

# llm.py:748
def _prepare_completion_params(self, messages, tools=None, skip_file_processing=False):
    ...
    params = {
        "model": self.model,
        "messages": formatted_messages,
        "temperature": self.temperature,
        "max_tokens": self.max_tokens or self.max_completion_tokens,
        "response_format": self.response_format,   # ★结构化输出：目标 Pydantic 类
        "stop": (self.stop_sequences or None) if self.supports_stop_words() else None,
        "tools": tools,                            # ★函数调用：工具 schema 列表
        "reasoning_effort": self.reasoning_effort,
        **self.additional_params,
    }
    return {k: v for k, v in params.items() if v is not None}   # ★去掉所有 None

模型返回 tool_call 后，这个方法负责"解析参数 + 执行函数"（llm.py:1734）：

# llm.py:1734
def _handle_tool_call(self, tool_calls, available_functions=None, from_task=None, from_agent=None):
    if not tool_calls or not available_functions:
        return None
    tool_call = tool_calls[0]                                    # ★只取第一个
    function_name = sanitize_tool_name(tool_call.function.name)
    function_args = {}
    if function_name in available_functions:
        try:
            function_args = json.loads(tool_call.function.arguments)  # ① 解析参数 JSON
            fn = available_functions[function_name]                   # ② 名字换回真函数
            crewai_event_bus.emit(self, event=ToolUsageStartedEvent(...))  # ③ 广播开始
            result = fn(**function_args)                              # ④ ★真正执行
            crewai_event_bus.emit(self, event=ToolUsageFinishedEvent(output=result, ...))
            self._handle_emit_call_events(response=result, call_type=LLMCallType.TOOL_CALL, ...)
            return result                                            # ⑤ 返回结果
        except Exception as e:
            logging.error(f"Error executing function '{function_name}': {e}")
            crewai_event_bus.emit(self, event=LLMCallFailedEvent(...))
            crewai_event_bus.emit(self, event=ToolUsageErrorEvent(...))   # 失败也广播
    return None

结构化输出的字段类型很讲究（llm.py:379，也见 base_llm.py:69）：

# llm.py:379
response_format: JsonResponseFormat | type[BaseModel] | None = None

# base_llm.py:69 —— JsonResponseFormat 是一个 TypedDict
class JsonResponseFormat(TypedDict):
    type: str            # 通常 "json_object" / "json_schema"
    ...

模型吐回来的文本，要经过这道校验才能变成对象（base_llm.py:897）：

# base_llm.py:897（配合 :77 的 _JSON_EXTRACTION_PATTERN = re.compile(r"\{.*}", re.DOTALL)）
def _validate_structured_output(response, response_format):
    if response_format is None:
        return response                              # 没要求结构化 → 原样返回
    try:
        if response.strip().startswith("{") or response.strip().startswith("["):
            data = json.loads(response)              # ① 整段就是 JSON → 直接 parse
            return response_format.model_validate(data)
        json_match = _JSON_EXTRACTION_PATTERN.search(response)   # ② 从夹带文字里抠 JSON
        if json_match:
            data = json.loads(json_match.group())
            return response_format.model_validate(data)
        raise ValueError("No JSON found in response")            # ③ 压根没 JSON
    except (json.JSONDecodeError, ValueError) as e:
        logging.warning(f"Failed to parse structured output: {e}")
        raise ValueError(f"Failed to parse response into {response_format.__name__}: {e}") from e

回看 L04 的 except 分支和 base_llm 里的执行封装（base_llm.py:714）：

# base_llm.py:714
def _handle_tool_execution(self, function_name, function_args, available_functions,
                           from_task=None, from_agent=None) -> str | None:
    """Handle tool execution with proper event emission."""
    ...  # 执行工具，并把开始/完成/错误统一包装成事件

# llm.py:1734 里工具执行失败时的处理（L04 的 except）
except Exception as e:
    fn = available_functions.get(function_name, lambda: None)
    logging.error(f"Error executing function '{function_name}': {e}")
    crewai_event_bus.emit(self, event=LLMCallFailedEvent(
        error=f"Tool execution error: {e!s}", from_task=from_task, from_agent=from_agent,
        call_id=get_current_call_id()))
    crewai_event_bus.emit(self, event=ToolUsageErrorEvent(
        tool_name=function_name, tool_args=function_args,
        error=f"Tool execution error: {e!s}", ...))