MindAgent

MindAgent 是一个基于 Python asyncio 的 Agent Runtime：用状态机约束生命周期，以 ReAct 循环（Think → Validate → Execute → Observe → Update）驱动模型决策，并将 Provider、Tool 与上下文管理解耦。

目前主流的Agent的核心，说起来就是Prompt（上下文构造）+ Loop（ReAct/状态机循环）+ Tool Call（统一为 Action 执行），在 MindAgent 中分别对应 ContextManager / ReActLoop+状态机 / ToolExecutor+ToolRegistry。

仓库：https://github.com/SyJarvis/mindagent.git

特性

• ReAct 闭环：Think → Validate → Execute → Observe → Update，单 Action 也走 ActionBatch(size=1)，每个 Action 只产生一个 Observation
• 受限并行：ActionScheduler 按 concurrency_key 与风险等级调度，READ_ONLY 可并行，WRITE / DANGEROUS / SEQUENTIAL 串行
• 三层超时与取消：Action / Batch / Run 超时，runtime.cancel(run_id) 直接中断并回收子任务
• OpenAI-compatible Provider：ProviderRouter 做能力路由，ProviderReasoner 统一决策
• Tool 体系：ToolRegistry + ToolExecutor，JSON Schema 参数校验，支持普通工具与流式工具
• 进程执行：短命令同步返回，长命令转为可轮询、可交互、可终止的 Process Session
• 上下文管理：ContextManager + ContextStore，tool call / result 原子 Bundle，压力水位与 Epoch 增量编译
• 会话边界：AgentSession（CommandQueue / EventQueue / SessionEnvironment / SessionPolicy），多轮隔离、workspace 与策略边界
• 可恢复执行：RunOutcome.PARTIAL + 确定性 checkpoint，CONTINUATION_REQUIRED 后经 continue_run() 跨 Run 继续
• 任务持久化：TaskCoordinator + LocalTaskLedgerStore + LocalTaskMemoryStore，进程重启后语义恢复
• 有界编排：AgentOrchestrator / AgentPipeline，限制并发、深度、子任务数与总超时，父取消级联子任务
• 可观测：TraceRecorder 输出 JSONL，可离线回放
• 幂等关闭：AgentRuntime.close() / AgentOrchestrator.close() 支持 async with 与依赖级联关闭

架构

AgentSession 边界：CommandQueue / EventQueue / SessionEnvironment / SessionPolicy
  │  create_session / run / run_context / close
  ▼
AgentRuntime ──► ReActLoop（Think → Validate → Execute → Observe → Update）
  │                 ├─► ProviderReasoner + ProviderRouter → ProviderResponse（统一决策）
  │                 ├─► PolicyEngine / CompletionGate（校验与完成门禁）
  │                 ├─► ActionScheduler → ActionDispatcher → ToolExecutor → ToolRegistry
  │                 │     （BaseTool 只收参数 + ToolContext；workspace / exec / memory tools）
  │                 └─► ContextManager + ContextStore（tool call / result 原子 Bundle）
横切：EventHandler / TraceRecorder（JSONL 落盘与回放）+ Heartbeat + Action / Batch / Run 三层超时
上层：TaskCoordinator + LocalTaskLedgerStore + LocalTaskMemoryStore（跨进程恢复）
编排：AgentOrchestrator / AgentPipeline（有界并发、深度、子任务数与总超时）

• core 只做编排，不直接依赖具体模型与工具：模型经 ProviderReasoner 收敛为 ProviderResponse，工具经 ToolExecutor + ToolRegistry 接入。
• 单 Action 也走 ActionBatch(size=1)，每个 Action 只产生一个 Observation，再聚合成 ObservationBatch。
• ActionScheduler 按 concurrency_key 与风险等级调度：READ_ONLY 可并行，WRITE / DANGEROUS / SEQUENTIAL 串行。
• 并行上限由 RunConfig.max_parallel_actions（默认 4）与 SessionPolicy.max_parallel_actions（默认 1）共同收敛。
• AgentSession 是外层边界：同一 Session 同时只跑一个 Run，不同 Session 可并发；TraceRecorder 只是 event_handler 的一种实现。
• workspace 文件工具与 exec 工具组使用 WorkspacePathResolver 校验路径或工作目录，但这不是 OS 级沙箱。

快速开始

完整可运行示例见 examples/tool_agent.py：

PYTHONPATH=src python examples/tool_agent.py

核心流程（摘自该示例，与当前 API 一致）：

from mindagent.context import ContextConfig, ContextManager
from mindagent.core import AgentRuntime, RunConfig
from mindagent.providers import ProviderReasoner, ProviderRouter
from mindagent.providers.openai import OpenAIProvider, OpenAIProviderParam
from mindagent.tools import BaseTool, ToolDefinition, ToolExecutor, ToolRegistry

registry = ToolRegistry([AddTool()])
provider = OpenAIProvider(OpenAIProviderParam.from_env())
reasoner = ProviderReasoner(
    ProviderRouter([provider]),
    tools=registry.provider_schemas(),
    action_risks=registry.action_risks(),
)
runtime = AgentRuntime(
    reasoner,
    ToolExecutor(registry),
    context_manager=ContextManager(
        ContextConfig(system_prompt="Use tools when required.")
    ),
    config=RunConfig(max_steps=4, step_timeout_s=60, total_timeout_s=120),
)

result = await runtime.run("Use the add tool to calculate 17 + 25.")
print(result.final_answer)

async with runtime:  # 或 await runtime.close()，幂等
    pass

自定义工具只需继承 BaseTool：

class AddTool(BaseTool):
    definition = ToolDefinition(
        name="add",
        description="Add two integers.",
        parameters={
            "type": "object",
            "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
            "required": ["a", "b"],
            "additionalProperties": False,
        },
    )

    async def execute(self, arguments, context):
        return arguments["a"] + arguments["b"]

安装

环境要求：Python 3.10+。

pip install mindagent
pip install "mindagent[openai]"

从源码开发：

uv pip install -e ".[openai]"

配置

在项目根目录创建 .env（勿提交真实密钥）：

MINDAGENT_API_KEY="your-api-key"
MINDAGENT_BASE_URL="https://your-provider.example/v1"
MINDAGENT_MODELS="your-model-name"
MINDAGENT_PLATFORM="openai-compatible"

OpenAIProviderParam.from_env() 自动读取以上变量，MINDAGENT_MODELS 支持逗号分隔，默认取第一个。

变量	说明
MINDAGENT_API_KEY	模型服务密钥
MINDAGENT_BASE_URL	OpenAI-compatible 服务地址
MINDAGENT_MODELS	模型名，逗号分隔，默认取第一个
MINDAGENT_PLATFORM	平台标识，当前为 openai-compatible

仅使用离线示例（如 bounded_multi_agent.py、runtime_resource_cleanup.py）时不需要配置密钥。

核心用法

Runtime：单次有界运行

AgentRuntime.run() 适用于无状态单任务。RunConfig 控制 max_steps、max_parallel_actions（默认 4）、action_timeout_s / batch_timeout_s / 总超时。超时或步数耗尽返回 RunOutcome.PARTIAL 而非直接报错。

Session：多轮会话边界

Session 是 ReAct 外层的上下文与策略边界：持有多轮消息、artifact、workspace、环境变量与安全策略。同一 Session 同时只跑一个 Run，不同 Session 可并发。

from pathlib import Path
from mindagent.core import SessionEnvironment, SessionEventType, SessionPolicy

session = runtime.create_session(
    environment=SessionEnvironment(
        workspace_root=Path.cwd(),
        env_vars={"PROJECT_ENV": "development"},
        memory_namespace="project-a",
    ),
    policy=SessionPolicy(allow_write=False, allow_dangerous=False),
)
await session.submit_run("分析当前项目结构")

while True:
    event = await session.next_event()
    if event.event_type == SessionEventType.RUN_COMPLETED:
        print(event.payload["result"].final_answer)
        break

await runtime.close_session(session.session_id)

Session 内并行度由 ActionScheduler 与 SessionPolicy.max_parallel_actions（默认 1）共同决定。

Workspace 与进程工具

内置文件工具是 read / write / edit，任务恢复配套工具是 memory_read。可执行工具系统由 command / poll / write_stdin / close_stdin / kill_session 五个 Agent Tool 组成，必须共享同一个 Session Manager：

from mindagent.tools import (
    FileEditTool,
    FileReadTool,
    FileWriteTool,
    ToolRegistry,
    create_exec_tools,
)

registry = ToolRegistry([
    FileReadTool(workspace),
    FileWriteTool(workspace),
    FileEditTool(workspace),
    *create_exec_tools(workspace),
])

command 只接受 shell string，可直接使用管道、重定向和条件执行；不接受命令数组以及旧版的超时/输出参数。短命令直接返回终态；超过 yield_time_ms 等待窗口的命令返回 session_id 并继续运行。完整工作流见使用 workspace 工具，精确字段见内置工具。示例：

PYTHONPATH=src python examples/file_tools_agent.py
PYTHONPATH=src python examples/coding_agent_loop.py

Trace：事件与回放

通过 event_handler 观察 STATE_CHANGED / DECISION_CREATED / ACTION_STARTED / ACTION_FINISHED / OBSERVATION_CREATED / FINAL_CREATED / HEARTBEAT 等事件。用 TraceRecorder 落盘为 JSONL：

from mindagent.core import TraceRecorder

recorder = TraceRecorder("traces")
runtime = AgentRuntime(reasoner, executor, event_handler=recorder)
events = TraceRecorder.replay("traces/<run_id>.jsonl")

Continuation：跨 Run 继续

max steps / step timeout / total timeout 触发时返回 PARTIAL 与内存 checkpoint，Session 发出 CONTINUATION_REQUIRED，确认后调用 continue_run() 新起 Run（继承 TaskState、Evidence、消息与 artifact，不恢复旧 ReActLoop）：

from mindagent.core import SessionEventType

event = await session.next_event()
if event.event_type == SessionEventType.CONTINUATION_REQUIRED:
    await session.continue_run(event.payload["checkpoint_id"])

Task：进程重启后恢复

聊天记录只恢复对话（LocalConversationStore，过滤 ToolCall / checkpoint 内部消息）；任务目标与进度恢复走 TaskCoordinator：

from mindagent.core import LocalTaskLedgerStore, LocalTaskMemoryStore, TaskCoordinator

coordinator = TaskCoordinator(
    session,
    LocalTaskLedgerStore(".mindagent/tasks"),
    LocalTaskMemoryStore(".mindagent/tasks/task-001"),
    task_id="task-001",
)
await coordinator.run("分析并修复当前项目")
# 重启后重建 Runtime / Session / Coordinator，再执行：
await coordinator.resume()

Ledger 只记目标、workspace、revision 与最近 Run 状态；压缩记忆存 memory.md，原始事件追加 events.jsonl，Agent 按需经 memory_read 读取。

应用

• mindcode（app/mindcode）：交互式 Coding Agent，任务入口已接入 TaskCoordinator：

  mindcode task run "分析并修复当前项目"
  mindcode task resume <task-id>
  

• mindwatch（app/mindwatch）：运行观测与回放配套应用。

两者独立测试、独立打包，核心协议由本仓库提供。

示例一览

examples/ 共 12 个可运行示例（均以 PYTHONPATH=src python examples/<name>.py 运行），按主题精选：

示例	说明
tool_agent.py	最小 Tool Calling 闭环（本 README 快速开始即摘自此）
coding_agent_loop.py	交互式 Coding Agent，多轮历史 + 串行 workspace 工具，WRITE / DANGEROUS 前终端确认
file_tools_agent.py	workspace 文件读取与五个进程工具的真实检索演示
context_manager.py	上下文构造、图像注入与裁剪行为
runtime_continuation_task.py	PARTIAL + CONTINUATION_REQUIRED + continue_run() 跨 Run 继续
runtime_resource_cleanup.py	Runtime / Orchestrator 幂等关闭与依赖级联清理
bounded_multi_agent.py	有界多 Agent 编排验收（并发 / 深度 / 超时上限）
agent_orchestrator_pipeline.py	AgentPipeline 声明式 DAG 包装
agent_orchestrator_openai.py	接入真实 OpenAI-compatible Provider 的多 Agent 编排
master_worker_orchestration.py	主从任务分解与执行树查询
model_dispatch_completion.py	ActionType.MODEL 模型动作分发（如图像理解）
orchestration_judge_prompt.py	编排裁决提示词构造

测试

make test            # 三套全部运行
make test-mindagent  # 本仓库：python -m pytest -q
make test-mindcode   # app/mindcode 内 pytest
make test-mindwatch  # app/mindwatch 内 pytest

三套测试相互独立：test-mindcode / test-mindwatch 会进入各自应用目录运行，避免不同应用的同名测试模块被 pytest 混合收集。

目录结构

src/mindagent/
  core/       状态机、ReActLoop、Runtime、Session、编排、Task、Trace
  context/    上下文构造、图像注入与裁剪
  providers/  Provider、Router、Reasoner（含 openai）
  tools/      Tool、Registry、Executor、workspace 内置工具
examples/     12 个可运行示例（tool_agent / coding_agent_loop / file_tools_agent / bounded_multi_agent / runtime_continuation_task 等）
tests/        核心单元测试
app/mindcode    编程助手应用
app/mindwatch   观测应用

路线图与非目标

近期方向以 CHANGELOG.md 为准。明确的非目标：

• OS 级沙箱 / 容器隔离（Session workspace 只是应用层路径边界）
• MCP / Skill Runtime
• 内建人工确认流程（策略拒绝后由宿主在 SessionPolicy 之外完成界面、审计与身份校验；Runtime 不保存确认状态）
• 自动 compaction（仅提供压力水位、Epoch 与 memory_read 按需读取）
• TUI / Web / RPC 传输层

贡献

欢迎提交 Issue 与 PR。请保持单 PR 只解决一个问题，附复现步骤与测试结果；新增行为请补充 tests/ 用例并跑通 make test-mindagent。

提交前请确认：未改动无关文件、无临时调试代码、无密钥落盘；Commit Message 只描述变更原因、方案与验证结果。

许可证

MIT，见 LICENSE。pyproject.toml 中的 license = "MIT" 与该文件一致。