> ## Documentation Index
> Fetch the complete documentation index at: https://qitor.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Agent Module

> AgentModule 是策略层，用来定义智能体如何初始化状态、构造提示词、做决策，并把观测结果归约到下一步状态。

`AgentModule` 是你定义智能体行为时要实现的核心契约。它负责智能体如何思考的全部内容：状态初始化、提示词构造、决策逻辑以及状态归约，但它**不执行循环本身**。执行循环由 [Engine](/zh/concepts/engine)（执行内核）负责。

## 泛型参数

`AgentModule` 有三个泛型类型参数：

```python theme={null}
class AgentModule(ABC, Generic[StateT, ObservationT, ActionT]):
    ...
```

| 参数             | 含义                                     |
| -------------- | -------------------------------------- |
| `StateT`       | 你的强类型状态类，必须继承 `StateSchema`            |
| `ObservationT` | Engine 传给各个钩子的观测结果（每步后智能体接收的结构化观察结果）类型 |
| `ActionT`      | `decide` 产生的动作（标准化工具调用）类型              |

一个完整的强类型声明通常长这样：

```python theme={null}
class MyAgent(AgentModule[MyState, dict[str, Any], Action]):
    ...
```

## 六个钩子

`AgentModule` 定义了六个钩子。其中两个是**必需的**：`init_state` 与 `reduce`。其余钩子都是**可选的**，并且有合理默认值。

### `init_state`（必需）

每次运行开始时调用一次。你需要返回一个初始状态对象，通常基于任务字符串初始化。

```python theme={null}
def init_state(self, task: str, **kwargs: Any) -> StateT:
    return MyState(task=task, max_steps=int(kwargs.get("max_steps", 10)))
```

所有传给 `Engine.run()` 或 `agent.run()` 的关键字参数都会流到这里，所以你可以通过 `kwargs` 接收 `max_steps` 或其他初始化字段。

### `reduce`（必需）

每次动作循环结束后调用。它把旧状态、新观测结果和本步决策归约成下一步状态——这是智能体从执行结果中学习的地方，也是归约的核心机制。

```python theme={null}
def reduce(
    self,
    state: StateT,
    observation: ObservationT,
    decision: Decision[ActionT],
) -> StateT:
    ...
    return state
```

<Warning>
  `reduce` 必须返回 `state` 或一个新的状态对象。返回 `None` 会触发运行时错误。
</Warning>

### `build_system_prompt`（可选）

返回动态系统提示词；如果返回 `None`，则本步不使用系统提示词。它会在每个决策步骤之前调用，因此提示词可以根据当前状态动态变化。

```python theme={null}
def build_system_prompt(self, state: StateT) -> str | None:
    return render_prompt(SYSTEM_TEMPLATE, {"tools": self.tool_registry.get_tool_descriptions()})
```

默认实现返回 `None`。

### `prepare`（可选）

把当前状态转成模型可直接消费的文本，也就是用户轮次。默认行为是 `str(state)`。通常你会覆写它，把多个状态字段组织成结构化输入。

```python theme={null}
def prepare(self, state: StateT) -> str:
    return f"Task: {state.task}\nStep: {state.current_step}/{state.max_steps}"
```

### `decide`（可选）

如果你想在 Engine 走默认大模型调用路径之前插入自定义决策逻辑，就覆写它。返回 `None` 时，Engine 会继续走模型调用与解析器路径。

```python theme={null}
def decide(self, state: StateT, observation: ObservationT) -> Decision[ActionT] | None:
    if state.current_step == 0:
        return Decision.wait("先检查当前编码上下文")
    return None
```

### `should_stop`（可选）

在每步 `reduce` 之后额外检查一次停止条件。返回 `True` 会终止运行。Engine 的内置停止条件仍会继续叠加生效。

```python theme={null}
def should_stop(self, state: StateT) -> bool:
    return state.final_result is not None
```

## 一个最小可运行示例

在 QitOS 里，公开的最小示例依然是一个真实的编码智能体：它会配置模型、挂载工作区，并让 Engine 驱动工具使用与验证。

```python theme={null}
from dataclasses import dataclass, field
from typing import Any

from qitos import Action, AgentModule, Decision, StateSchema
from qitos.kit import REACT_SYSTEM_PROMPT, ReActTextParser, format_action, render_prompt
from qitos.kit.toolset import coding_tools
from qitos.models import OpenAICompatibleModel


@dataclass
class MinimalCodingState(StateSchema):
    scratchpad: list[str] = field(default_factory=list)
    target_file: str = "buggy_module.py"
    test_command: str = 'python -c "import buggy_module; assert buggy_module.add(20, 22) == 42"'


class MinimalCodingAgent(AgentModule[MinimalCodingState, dict[str, Any], Action]):
    def __init__(self, llm: OpenAICompatibleModel, workspace_root: str):
        super().__init__(
            toolset=[coding_tools(workspace_root=workspace_root, shell_timeout=20, include_notebook=False)],
            llm=llm,
            model_parser=ReActTextParser(),
        )

    def init_state(self, task: str, **kwargs: Any) -> MinimalCodingState:
        return MinimalCodingState(
            task=task,
            max_steps=int(kwargs.get("max_steps", 8)),
            target_file=str(kwargs.get("target_file", "buggy_module.py")),
            test_command=str(kwargs.get("test_command")),
        )

    def build_system_prompt(self, state: MinimalCodingState) -> str | None:
        _ = state
        return render_prompt(
            REACT_SYSTEM_PROMPT,
            {"tool_schema": self.tool_registry.get_tool_descriptions()},
        )

    def reduce(
        self,
        state: MinimalCodingState,
        observation: dict[str, Any],
        decision: Decision[Action],
    ) -> MinimalCodingState:
        action_results = observation.get("action_results", [])
        if decision.rationale:
            state.scratchpad.append(f"Thought: {decision.rationale}")
        if decision.actions:
            state.scratchpad.append(f"Action: {format_action(decision.actions[0])}")
        if action_results:
            first = action_results[0]
            state.scratchpad.append(f"Observation: {first}")
            if isinstance(first, dict) and int(first.get("returncode", 1)) == 0:
                state.final_result = "Patch applied and verification passed."
        return state
```

## 如何运行智能体

调用 `agent.run()` 即可执行一个任务。这是最常用的入口。它会在内部创建 `Engine`、运行循环，并返回结果。

```python theme={null}
agent = MinimalCodingAgent(
    llm=build_model(),
    workspace_root="./playground/minimal_coding_agent",
)

# 默认只返回 final_result 字符串
answer = agent.run(
    "Fix the bug in buggy_module.py and make the verification command pass.",
    workspace="./playground/minimal_coding_agent",
    max_steps=8,
)

# return_state=True 时返回完整 EngineResult
result = agent.run(
    "Fix the bug in buggy_module.py and make the verification command pass.",
    workspace="./playground/minimal_coding_agent",
    max_steps=8,
    return_state=True,
)
print(result.state.final_result)
print(result.state.stop_reason)
```

`agent.run()` 最常见参数如下：

| 参数              | 类型                    | 说明                                             |
| --------------- | --------------------- | ---------------------------------------------- |
| `task`          | `str \| Task`         | 普通文本任务或结构化 `Task` 对象                           |
| `max_steps`     | `int \| None`         | 覆盖本次运行的步数预算                                    |
| `workspace`     | `str \| None`         | 挂载到宿主环境的工作目录                                   |
| `return_state`  | `bool`                | 返回 `EngineResult`，而不是 `state.final_result`     |
| `trace`         | `bool \| TraceWriter` | `True` 自动启用追踪记录，`False` 关闭，或直接传入 `TraceWriter` |
| `trace_logdir`  | `str`                 | 追踪记录输出目录，默认 `"./runs"`                         |
| `parser`        | `Parser`              | 与提示词格式匹配的输出解析器（将原始模型输出解析为 Decision 的组件）        |
| `critics`       | `list[Critic]`        | 每步执行后评估的评估器（可批准、停止或重试当前步骤）                     |
| `stop_criteria` | `list[StopCriteria]`  | 除默认条件外的附加停止条件                                  |

<Tip>
  如果你需要更细粒度的控制，例如自定义钩子、分支选择器，或在多次运行之间复用同一个 Engine，可以使用 `agent.build_engine(**kwargs)`，然后直接调用 `engine.run(task)`。
</Tip>

## 构造函数参数

`AgentModule.__init__` 支持这些参数，全部可选：

```python theme={null}
AgentModule(
    tool_registry=registry,   # ToolRegistry：智能体可用的工具
    llm=model,                # 大模型客户端，供 Engine 默认模型路径使用
    model_parser=parser,      # 把原始模型输出解析成 Decision 的解析器
    memory=memory,            # 跨步或跨运行的记忆实现
    history=history,          # 单次运行内的历史实现
    **config,                 # 额外配置，保存在 self.config
)
```