跳转到主要内容
AgentModule 是你定义 agent 行为时要实现的核心契约。它负责 agent “如何思考”的全部内容：状态初始化、prompt 构造、决策逻辑以及状态归约，但它不执行循环本身。执行循环由 Engine 负责。

​
泛型参数

AgentModule 有三个泛型类型参数： 
class AgentModule(ABC, Generic[StateT, ObservationT, ActionT]):
    ...
参数含义
StateT你的强类型状态类，必须继承 StateSchema
ObservationTEngine 传给各个 hooks 的 observation 类型
ActionTdecide 产生的 action 类型
一个完整的强类型声明通常长这样： 
class MyAgent(AgentModule[MyState, dict[str, Any], Action]):
    ...

​
六个 hooks

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

​
init_state（必需）

每次 run 开始时调用一次。你需要返回一个初始状态对象，通常基于 task 字符串初始化。 
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（必需）

每次 action cycle 结束后调用。它把“旧状态 + 新观测 + 本步决策”归约成“下一步状态”。这就是 agent 从执行结果中学习的地方。 
def reduce(
    self,
    state: StateT,
    observation: ObservationT,
    decision: Decision[ActionT],
) -> StateT:
    ...
    return state
reduce 必须返回 state 或一个新的 state 对象。返回 None 会触发运行时错误。

​
build_system_prompt（可选）

返回动态 system prompt；如果返回 None，则本步不使用 system prompt。它会在每个 decide step 之前调用，因此 prompt 可以根据当前 state 动态变化。 
def build_system_prompt(self, state: StateT) -> str | None:
    return render_prompt(SYSTEM_TEMPLATE, {"tools": self.tool_registry.get_tool_descriptions()})
默认实现返回 None。

​
prepare（可选）

把当前 state 转成模型可直接消费的文本，也就是 user turn。默认行为是 str(state)。通常你会覆写它，把多个状态字段组织成结构化输入。 
def prepare(self, state: StateT) -> str:
    return f"Task: {state.task}\nStep: {state.current_step}/{state.max_steps}"

​
decide（可选）

如果你想在 Engine 走默认 LLM 路径之前插入自定义决策逻辑，就覆写它。返回 None 时，Engine 会继续走模型调用与 parser 路径。 
def decide(self, state: StateT, observation: ObservationT) -> Decision[ActionT] | None:
    if state.current_step == 0:
        return Decision.wait("先检查当前 coding context")
    return None

​
should_stop（可选）

在每步 reduce 之后额外检查一次停止条件。返回 True 会终止运行。Engine 的内置 stop criteria 仍会继续叠加生效。 
def should_stop(self, state: StateT) -> bool:
    return state.final_result is not None

​
一个最小可运行示例

在 QitOS 里，公开的最小示例依然是一个真实 coding agent：它会配置模型、挂载 workspace，并让 Engine 驱动工具使用与验证。 
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

调用 agent.run() 即可执行一个任务。这是最常用入口。它会在内部创建 Engine、运行循环，并返回结果。 
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() 最常见参数如下：
参数类型说明
taskstr | Task普通文本任务或结构化 Task 对象
max_stepsint | None覆盖本次运行的步数预算
workspacestr | None挂载到 host environment 的工作目录
return_statebool返回 EngineResult，而不是 state.final_result
tracebool | TraceWriterTrue 自动启用 trace，False 关闭，或直接传入 TraceWriter
trace_logdirstrtrace 输出目录，默认 "./runs"
parserParser与 prompt 格式匹配的输出 parser
criticslist[Critic]每步执行后评估的 critics
stop_criterialist[StopCriteria]除默认条件外的附加停止条件
如果你需要更细粒度控制，例如自定义 hooks、branch selector，或在多次 run 之间复用同一个 Engine，可以使用 agent.build_engine(**kwargs)，然后直接调用 engine.run(task)。

​
构造函数参数

AgentModule.__init__ 支持这些参数，全部可选： 
AgentModule(
    tool_registry=registry,   # ToolRegistry：agent 可用的工具
    llm=model,                # LLM client，供 Engine 默认模型路径使用
    model_parser=parser,      # 把原始模型输出解析成 Decision 的 parser
    memory=memory,            # 跨步或跨 run 的 memory 实现
    history=history,          # 单次 run 内的 history 实现
    **config,                 # 额外配置，保存在 self.config
)