类的 docstring 直接引用了论文并说明了这个定位（step_executor.py:63）：

# step_executor.py:63（类 docstring 节选）
class StepExecutor:
    """Executes a SINGLE todo item using direct-action execution.
    The StepExecutor owns its own message list per invocation. It never reads
    or writes the AgentExecutor's state. Results flow back via StepResult.
    Execution pattern (per Plan-and-Act, arxiv 2503.09572): ...
        No inner loop — recovery is PlannerObserver's responsibility."""

__init__（step_executor.py:90）在构造时就把工具、原生支持检测好，避免每步重复判断：

# step_executor.py:90
def __init__(self, llm, tools, agent, original_tools=None, tools_handler=None,
             task=None, crew=None, function_calling_llm=None, ...):
    self.llm = llm
    self.tools = tools                       # 结构化工具（供文本模式描述用）
    self.original_tools = original_tools or []
    self.tools_handler = tools_handler       # 复用 Day 11 的缓存中枢
    ...
    self._use_native_tools = check_native_tool_support(   # ★构造时就定：走原生还是文本
        self.llm, self.original_tools)
    self._openai_tools: list[dict[str, Any]] = []
    self._available_functions: dict[str, Callable[..., Any]] = {}
    if self._use_native_tools and self.original_tools:
        (self._openai_tools, self._available_functions, _) = setup_native_tools(
            self.original_tools)             # 提前把工具转成 OpenAI schema

对外入口 execute（step_executor.py:126）：

# step_executor.py:126
def execute(self, todo: TodoItem, context: StepExecutionContext,
            max_step_iterations: int = 15, step_timeout: int | None = None) -> StepResult:
    start_time = time.monotonic()
    tool_calls_made: list[str] = []
    try:
        enforce_rpm_limit(self.request_within_rpm_limit)
        messages = self._build_isolated_messages(todo, context)   # ① 造独立消息
        if self._use_native_tools:
            result_text = self._execute_native(messages, todo, tool_calls_made, ...)   # ② 分流
        else:
            result_text = self._execute_text_parsed(messages, todo, tool_calls_made, ...)
        self._validate_expected_tool_usage(todo, tool_calls_made)   # ③ 校验该用的工具用了没
        elapsed = time.monotonic() - start_time
        return StepResult(success=True, result=result_text,
                          tool_calls_made=tool_calls_made, execution_time=elapsed)   # ④ 打包
    except Exception as e:
        ...   # 见 L07 的降级/失败处理
        return StepResult(success=False, result="", error=str(e), ...)

_build_isolated_messages（step_executor.py:233）为这一步现造消息：

# step_executor.py:233
def _build_isolated_messages(self, todo, context) -> list[LLMMessage]:
    system_prompt = self._build_system_prompt()          # 系统提示：你是执行者，只干这一步
    user_prompt = self._build_user_prompt(todo, context)  # 用户提示：步骤描述 + 依赖结果 + 工具
    return [
        format_message_for_llm(system_prompt, role="system"),
        format_message_for_llm(user_prompt, role="user"),
    ]

# step_executor.py:249 系统提示的组装
def _build_system_prompt(self) -> str:
    role = self.agent.role if self.agent else "Assistant"
    goal = self.agent.goal if self.agent else "Complete tasks efficiently"
    backstory = getattr(self.agent, "backstory", "") or ""
    ...
    return I18N_DEFAULT.retrieve("planning", "step_executor_system_prompt").format(
        role=role, backstory=backstory, goal=goal, tools_section=tools_section)

虽说是"单步"，但一步内部仍允许多轮（文本模式 _execute_text_parsed，step_executor.py:317）：

# step_executor.py:317
def _execute_text_parsed(self, messages, todo, tool_calls_made,
                         max_step_iterations=15, step_timeout=None, start_time=None) -> str:
    use_stop_words = self.llm.supports_stop_words() if self.llm else False
    last_tool_result = ""
    for _ in range(max_step_iterations):            # ★步内有上限的小循环
        if step_timeout and start_time:
            if time.monotonic() - start_time >= step_timeout:
                return last_tool_result or f"Step timed out after ..."   # 超时兜底
        answer = self.llm.call(messages, callbacks=self.callbacks, ...)
        if not answer:
            raise ValueError("Empty response from LLM")
        formatted = process_llm_response(str(answer), use_stop_words)   # 复用 Day 09 解析
        if isinstance(formatted, AgentFinish):
            return str(formatted.output)             # 给答案 → 这步完成
        if isinstance(formatted, AgentAction):
            tool_calls_made.append(formatted.tool)
            tool_result = self._execute_text_tool_with_events(formatted, todo)  # 执行工具+发事件
            last_tool_result = tool_result
            messages.append({"role": "assistant", "content": str(answer)})
            messages.append(self._build_observation_message(tool_result))       # 观察喂回
            continue
        return answer_str
    return last_tool_result

_build_observation_message（step_executor.py:474）有个巧妙设计——工具若返回图片，转成多模态消息：

# step_executor.py:474
@staticmethod
def _build_observation_message(tool_result: str) -> LLMMessage:
    parsed = StepExecutor._parse_vision_sentinel(tool_result)   # 解析 "VISION_IMAGE:..." 前缀
    if parsed:
        media_type, b64_data = parsed
        return {"role": "user", "content": [
            {"type": "text", "text": "Observation: Here is the image:"},
            {"type": "image_url",
             "image_url": {"url": f"data:{media_type};base64,{b64_data}"}},   # 转成 data URI
        ]}
    return {"role": "user", "content": f"Observation: {tool_result}"}          # 普通文本观察

# step_executor.py:462
@staticmethod
def _parse_vision_sentinel(raw: str) -> tuple[str, str] | None:
    prefix = "VISION_IMAGE:"
    if not raw.startswith(prefix):
        return None
    rest = raw[len(prefix):]
    sep = rest.find(":")
    if sep <= 0:
        return None
    return rest[:sep], rest[sep + 1:]      # 返回 (media_type, base64_data)

execute 的 except 分支里有一处精心设计的降级（step_executor.py:183）：

# step_executor.py:183
except Exception as e:
    if self._use_native_tools and is_native_tool_calling_unsupported_error(e):
        try:
            self._use_native_tools = False       # 关掉原生
            self._openai_tools = []; self._available_functions = {}
            # 保留已有对话（含已完成的原生工具往返），只追加文本工具说明，
            # 不重启这一步——避免已执行的工具被重复执行
            messages.append(format_message_for_llm(
                build_text_tool_calling_fallback_message(...), role="user"))
            result_text = self._execute_text_parsed(messages, todo, tool_calls_made, ...)
            ...
            return StepResult(success=True, result=result_text, ...)
        except Exception as fallback_error:
            e = fallback_error
    return StepResult(success=False, result="", error=str(e), ...)

成功路径最后还有一道强制校验 _validate_expected_tool_usage（step_executor.py:504）：

# step_executor.py:504
def _validate_expected_tool_usage(self, todo, tool_calls_made) -> None:
    expected_tool = getattr(todo, "tool_to_use", None)
    if not expected_tool:
        return                                   # 没指定必用工具 → 不校验
    expected_tool_name = sanitize_tool_name(expected_tool)
    available_tool_names = {...} | set(self._available_functions.keys())
    if expected_tool_name not in available_tool_names:
        return                                   # 指定的工具压根不可用 → 跳过
    called_names = {sanitize_tool_name(n) for n in tool_calls_made}
    if expected_tool_name not in called_names:
        raise ValueError(f"Expected tool '{expected_tool_name}' was not called ...")

P=lib/crewai/src/crewai
sed -n '63,124p'  $P/agents/step_executor.py    # 定位 + 初始化
sed -n '126,231p' $P/agents/step_executor.py    # execute + 降级 + StepResult
sed -n '317,367p' $P/agents/step_executor.py    # 文本步内小循环
sed -n '462,526p' $P/agents/step_executor.py    # 视觉哨兵 + 强制校验
grep -n "arxiv 2503.09572" $P/agents/step_executor.py   # Plan-and-Act 出处