Agent Module

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

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

​

泛型参数

AgentModule 有三个泛型类型参数：

class AgentModule(ABC, Generic[StateT, ObservationT, ActionT]):
    ...

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

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

class MyAgent(AgentModule[MyState, dict[str, Any], Action]):
    ...

​

六个钩子

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

​

init_state（必需）

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

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（必需）

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

def reduce(
    self,
    state: StateT,
    observation: ObservationT,
    decision: Decision[ActionT],
) -> StateT:
    ...
    return state

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

​

build_system_prompt（可选）

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

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)。通常你会覆写它，把多个状态字段组织成结构化输入。

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

​

decide（可选）

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

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 的内置停止条件仍会继续叠加生效。

def should_stop(self, state: StateT) -> bool:
    return state.final_result is not None

​

一个最小可运行示例

在 QitOS 里，公开的最小示例依然是一个真实的编码智能体：它会配置模型、挂载工作区，并让 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.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() 最常见参数如下：

参数	类型	说明
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]	除默认条件外的附加停止条件

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

​

构造函数参数

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

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