MindAgent

MindAgent 是一个基于 Python 和 asyncio 的 Agent Runtime。它使用状态机约束生命周期，通过 ReAct 循环驱动模型决策，并将 Provider、Tool 和上下文管理分离。

当前版本已完成单 Agent、多 Action 受限并行 Tool Calling 的核心闭环，并提供 Session Actor 作为多轮上下文、环境和安全策略的运行边界。

当前能力

• 状态机控制 Agent 生命周期
• ReAct：Think → Validate → Execute → Observe → Update
• DecisionKind / ActionType 分离
• ActionBatch / ObservationBatch 批次协议
• ActionScheduler 受限并行与稳定结果排序
• 批次部分失败和部分策略拒绝恢复
• Action/Batch/Run 三层超时与并行任务取消清理
• OpenAI-compatible Provider
• Provider 能力路由
• Tool 注册、JSON Schema 和参数校验
• 普通工具与流式工具
• ContextManager 文本、图像和 Tool Observation 管理
• 上下文长度估算与旧轮次裁剪
• 事件、heartbeat、超时、取消和人工审批
• 策略拒绝后的用户询问与 ReAct 恢复
• 结构化错误码、错误阶段和可恢复标记
• JSONL 执行轨迹与离线回放
• Session CommandQueue / EventQueue 通信
• Session 多轮上下文保持与会话间隔离
• Session workspace、环境变量、记忆命名空间和策略边界
• RunOutcome、ExecutionBoundary 和确定性 continuation checkpoint
• Max steps、Step timeout、Run timeout 后的 PARTIAL 结果
• Session CONTINUATION_REQUIRED 和用户确认后的跨 Run 继续
• 任务记忆原子持久化与重启后的 workspace 语义恢复
• AgentOrchestrator 注册多个 Agent、执行子 Agent 任务并查询执行树/图
• AgentPipeline 作为 AgentOrchestrator 的声明式 DAG 包装
• 内置时间、计算器、上下文查询、进程内记忆和图像理解工具

MCP、Skill、长期记忆、跨进程子 Agent 管理和完整多 Agent 自治编排尚未实现。

架构

AgentRuntime
  ├── AgentOrchestrator
  │     ├── AgentWorkerSpec
  │     ├── SubAgentExecutor → AGENT Action
  │     └── AgentPipeline
  ├── AgentSession
  │     ├── CommandQueue
  │     ├── EventQueue
  │     ├── SessionEnvironment
  │     └── SessionPolicy
  ├── ReActLoop
  │     ├── ProviderReasoner
  │     │     └── ProviderRouter → LLM/VLM Provider
  │     ├── PolicyEngine
  │     ├── ActionScheduler
  │     │     └── ActionDispatcher
  │     │           └── TOOL → ToolExecutor → ToolRegistry → Tool
  │     └── ContextManager
  ├── EventHandler
  ├── Heartbeat
  └── Timeout / Cancel

核心边界：

• core 只负责编排和协议，不依赖具体模型或工具。
• providers 将不同模型接口统一为 ProviderResponse。
• tools 不感知 Agent 推理，只接收参数和 ToolContext。
• 所有外部动作通过带稳定 action_id 的 ActionRequest 表达。
• 单 Action 也使用 ActionBatch(size=1)；每个 Action 最终只生成一个 Observation。
• ActionDispatcher 按 ActionType 路由单 Action Executor，未注册类型返回 结构化错误。
• ActionScheduler 并行执行无资源冲突的 READ_ONLY Action；相同 concurrency_key、WRITE、DANGEROUS 和 SEQUENTIAL batch 保持串行。
• RunConfig.max_parallel_actions 默认是 4；Session 使用 SessionPolicy.max_parallel_actions 进一步收紧，默认是 1。
• RunConfig.action_timeout_s 和 batch_timeout_s 分别控制单 Action 与整个 batch；runtime.cancel(run_id) 会取消并回收当前批次全部子任务。

进程内多 Agent 编排默认有界。AgentOrchestrator 可限制全局任务并发、后台任务数、 任务树深度、单父任务累计子任务数和任务总超时。取消父任务会级联取消运行中的后代； 父任务进入终态时，仍由它拥有的后台后代会在父任务返回前被取消并等待结束。

orchestrator = AgentOrchestrator(
    max_concurrent_tasks=8,
    max_background_tasks=8,
    max_task_depth=4,
    max_children_per_task=8,
    task_timeout_s=300,
)

离线有界编排验收：

PYTHONPATH=src python examples/bounded_multi_agent.py

资源生命周期

Runtime 和 Orchestrator 都支持幂等异步关闭及 async with：

async with AgentRuntime(reasoner, executor) as runtime:
    result = await runtime.run("执行一个有界任务")

AgentRuntime.close() 会关闭 Session、取消并等待活跃 Run。Reasoner 和 Executor 属于注入依赖，默认不关闭；由 Runtime 独占时使用 close(close_dependencies=True) 级联关闭 Provider、Executor 和 Tool。

AgentOrchestrator.close() 会取消并等待前台、后台 Agent task，并终止事件订阅。 注册的 worker Runtime 默认由调用方管理；使用 close(close_workers=True) 显式转移 关闭责任。关闭开始后，Runtime 和 Orchestrator 都拒绝启动新工作。

离线资源清理验收：

PYTHONPATH=src python examples/runtime_resource_cleanup.py

环境要求

• Python 3.10+
• OpenAI-compatible API（仅使用 OpenAI Provider 时需要）

安装核心 Runtime：

pip install mindagent

使用 OpenAI-compatible Provider：

pip install "mindagent[openai]"

从源码开发：

uv pip install --python .venv/bin/python -e ".[openai]"

API 配置

在项目根目录创建 .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 支持逗号分隔，当前默认使用第一个模型。

不要提交包含真实密钥的 .env。

快速开始

运行完整 Tool Calling 示例：

PYTHONPATH=src .venv/bin/python examples/tool_agent.py

示例包含：

1. 定义一个 AddTool
2. 注册到 ToolRegistry
3. 将工具 Schema 提供给模型
4. 创建 Provider、ContextManager 和 Runtime
5. 模型自主调用工具
6. Observation 写回上下文
7. 模型生成最终答案

核心代码：

registry = ToolRegistry([AddTool()])
provider = OpenAIProvider(OpenAIProviderParam.from_env())

reasoner = ProviderReasoner(
    ProviderRouter([provider]),
    tools=registry.provider_schemas(),
    action_risks=registry.action_risks(),
    approval_required=registry.approval_required(),
)

runtime = AgentRuntime(
    reasoner,
    ToolExecutor(registry),
    context_manager=ContextManager(
        ContextConfig(system_prompt="Use tools when required.")
    ),
)

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

完整代码见 examples/tool_agent.py。

Session

一个 Session 是 ReActLoop 外层的上下文环境。它负责持有多轮消息、artifact、 workspace、环境变量和安全策略，通过 CommandQueue 接收控制命令，通过 EventQueue 输出运行事件。

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:
        result = event.payload["result"]
        print(result.final_answer)
        break

await runtime.close_session(session.session_id)

同一 Session 同时只运行一个 Run，保证消息和 artifact 更新顺序确定；不同 Session 可以由同一个 Runtime 并发运行。Session 内部 Action 是否并行由 ActionScheduler 和 SessionPolicy.max_parallel_actions 共同决定。

Session workspace 是应用层路径边界，不等同于操作系统容器或进程沙箱。自定义 Tool 应通过 ToolContext 读取 Session 环境和策略，避免使用进程级可变全局状态。 allow_network 已进入策略模型，但要等网络类 Executor 接入后才能统一执行校验。

当 Run 达到 max steps、step timeout 或 total timeout 时，Runtime 会返回 RunOutcome.PARTIAL 和确定性 checkpoint。Session 随后发送 CONTINUATION_REQUIRED；调用方确认后使用 continue_run() 创建新的 Run：

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

新 Run 继承 TaskState、Evidence、Session messages 和 artifacts，但使用新的 run_id，不会恢复旧 ReActLoop。当前 checkpoint 只保存在 Session 内存中，不支持 进程重启恢复。

如果只需要在进程重启后恢复聊天记录，可以配置 LocalConversationStore：

from mindagent.core import LocalConversationStore

conversation_store = LocalConversationStore(
    ".mindagent/conversations"
)
session = runtime.create_session(
    session_id="task-session",
    conversation_store=conversation_store,
)

它只保存普通 user/assistant 消息，并过滤 ToolCall、Tool Result、checkpoint 和 CompletionGate 内部反馈。重启后恢复的消息只作为新 Run 的历史对话，不恢复 TaskState、Evidence、Artifact、旧 run_id 或 pending continuation。

如果需要恢复长期任务目标和进度，在 Session 外层使用 TaskCoordinator 和单写者 Task Ledger：

from mindagent.core import (
    LocalTaskLedgerStore,
    LocalTaskMemoryStore,
    TaskCoordinator,
)

memory_store = LocalTaskMemoryStore(
    ".mindagent/tasks/task-001"
)
session = runtime.create_session(
    session_id="task-session",
    environment=SessionEnvironment(
        workspace_root=Path.cwd(),
    ),
    conversation_store=conversation_store,
)
coordinator = TaskCoordinator(
    session,
    LocalTaskLedgerStore(".mindagent/tasks"),
    memory_store,
    task_id="task-001",
)

# 第一次执行；TaskCoordinator 在 Run 启动前保存 active 状态。
await coordinator.run("分析并修复当前项目")

# 进程重启后重新创建 Runtime、Session 和 Coordinator。
await coordinator.resume()

Task Ledger 只记录目标、workspace、revision 和最近 Run 状态。压缩后的工作记忆 保存在同一任务目录的 memory.md，原始运行事件追加到 events.jsonl。 resume() 会在线程外采集当前 workspace 顶层内容、有限 Git status 和 Git HEAD， 再通过结构化 system/task context 创建新 Run。它保留原始 TaskState.objective， 不会把恢复快照写入普通对话历史，也不会重放旧 Action。历史内容不会自动全量 注入；Agent 需要时通过 memory_read 读取 memory.md 或近期事件。 LocalTaskMemoryStore.replace() 是唯一的记忆正文写入路径：主 Agent可以调度 子 Agent压缩旧记忆与近期事件，审核其输出后再原子替换 memory.md。

mindcode 的任务入口已接入同一链路：

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

内置工具

from mindagent.tools import (
    CalculatorTool,
    ContextQueryTool,
    ImageUnderstandingTool,
    MemoryTool,
    MemoryReadTool,
    TimeNowTool,
)

registry = ToolRegistry(
    [
        TimeNowTool(),
        CalculatorTool(),
        ContextQueryTool(),
        MemoryTool(),
        MemoryReadTool(memory_store),
    ]
)

• time_now：获取指定 IANA 时区的当前时间
• calculator：基于受限 AST 计算算术表达式
• context_query：读取当前 run 的 metadata 或 artifacts
• memory：按 session/run namespace 隔离的进程内 key/value 存储
• memory_read：按需读取任务的磁盘 memory.md 或近期 events.jsonl
• image_understanding：通过 ProviderRouter 选择多模态 provider 分析图像

ImageUnderstandingTool 需要显式传入 router：

image_tool = ImageUnderstandingTool(
    ProviderRouter([vision_provider])
)

它支持 HTTP 图片 URL 和 base64 data URL。默认 api="auto"，优先使用 Chat Completions 的 text / image_url 格式；当兼容服务明确返回端点或 图像能力不支持的 4xx 时，再回退 Responses API 的 input_text / input_image 格式。也可以显式设置：

arguments = {
    "image_url": image_data_url,
    "prompt": "请描述图片。",
    "api": "chat_completions",  # 或 "responses"
}

定义工具

普通工具继承 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"]

流式工具继承 StreamingTool，并产生 ToolProgress：

async def stream(self, arguments, context):
    yield ToolProgress(message="working", progress=0.5)
    yield ToolProgress(message="done", data=result, progress=1.0)

进度通过 ACTION_PROGRESS 事件发送。所有 chunk 会由 ToolExecutor 聚合，ReAct 循环只接收一个最终 Observation。

Workspace 工具

MindAgent 提供四个受 workspace 边界限制的基础工具：

• file_search：搜索文件名或文本内容
• file_read：按行号或 anchor 提取文本片段
• file_edit：创建、覆盖或唯一文本替换
• exec_command：以 argv 数组执行命令，不启用 shell

registry = ToolRegistry(
    [
        FileSearchTool(workspace),
        FileReadTool(workspace),
        FileEditTool(workspace),
        ExecCommandTool(workspace),
    ]
)

路径会经过 WorkspacePathResolver 校验，不能通过绝对路径、.. 或 symlink 访问 workspace 外部。exec_command 根据 argv 动态判断风险： 只读命令为 READ_ONLY，普通命令为 WRITE，危险命令为 DANGEROUS。

真实文件检索示例：

PYTHONPATH=src .venv/bin/python examples/file_tools_agent.py

交互式 Coding Agent 示例：

PYTHONPATH=src .venv/bin/python examples/coding_agent_loop.py

该示例保留多轮对话历史，串行执行 workspace 工具，并在执行 WRITE 或 DANGEROUS 操作前请求终端确认。输入 /clear 清空历史， 输入 /exit 退出。

ContextManager

ContextManager 当前负责：

• 以只追加 ContextStore 保存原始执行记录
• 将 tool call 与 tool result 组成不可拆分的原子 Bundle
• 注入 system、user 和多模态图片内容
• 每次 THINK 前动态编译 Provider 消息视图
• 计算稳定前缀指纹，保持 system 与工具 schema 顺序稳定
• 使用 NORMAL / WARNING / CRITICAL 压力水位决定是否压缩
• 通过 Context Epoch 避免 ReAct iteration 内反复重写历史
• 通过 CONTEXT_PACKED 事件记录预算和压缩决策

示例：

result = await runtime.run(
    "Describe this image.",
    artifacts={
        "images": ["https://example.com/image.png"],
    },
)

Context 不使用固定分块预算。默认使用离线字符估算，避免初始化时下载 tokenizer；需要精确估算时可设置 ContextConfig(use_tiktoken=True)。

事件

通过 event_handler 观察执行过程：

async def handle_event(event):
    print(event.event_type.value, event.payload)

runtime = AgentRuntime(
    reasoner,
    executor,
    event_handler=handle_event,
)

主要事件包括：

• STATE_CHANGED
• DECISION_CREATED
• ACTION_VALIDATED
• ACTION_STARTED
• ACTION_PROGRESS
• ACTION_FINISHED
• OBSERVATION_CREATED
• USER_INPUT_REQUESTED
• USER_INPUT_RECEIVED
• FINAL_CREATED
• ERROR_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")

runtime.cancel(run_id) 会直接中断当前 LLM 或工具 await，并返回 CANCELLED 状态；它不依赖 step_timeout_s。

策略拒绝恢复

策略拒绝 Action 时，Runtime 不会直接进入 ERROR。它会先写入一个失败 Observation，保持 assistant(tool_calls) 与 tool 消息配对，然后进入 WAITING_FOR_USER：

task = asyncio.create_task(
    runtime.run("update the file", run_id="run-1")
)

while True:
    status = runtime.get_run_status("run-1")
    if status and status.state == AgentState.WAITING_FOR_USER:
        print(status.pending_user_question)
        runtime.respond_to_user(
            "run-1",
            "不要写文件，改用只读方式继续。",
        )
        break
    await asyncio.sleep(0.05)

result = await task

恢复路径为：

ACTION_VALIDATING
  → OBSERVING
  → CONTEXT_UPDATING
  → WAITING_FOR_USER
  → THINKING

AgentResult.error_info 和 Observation.error_info 提供结构化错误： code、phase、recoverable 和 details。原有字符串 error 字段继续 保留，兼容现有调用方。

Ask User 与 Human Approval

ASK_USER 用于让 LLM 主动补充任务信息，例如澄清目标、确认偏好或选择方案。 HumanApprovalHandler 用于 Runtime/Policy 拦截具体 Action 的副作用风险，例如 WRITE、DANGEROUS 或外部系统变更。LLM 可以表达风险意见，但不能自行决定是否 绕过审批；最终放行由 Policy 和 Runtime 控制。

审批 handler 可以继续返回 bool，也可以返回带反馈的 ApprovalResult。当用户 拒绝审批并提供反馈时，Runtime 会把反馈写入失败 Observation，使后续 ReAct 循环 可以基于该反馈调整策略。

测试

在仓库根目录运行全部独立测试套件：

make test

也可以单独运行一个项目，避免不同应用的同名测试模块被 pytest 混合收集：

make test-mindagent
make test-mindcode
make test-mindwatch

MindAgent 核心的直接命令为：

python -m pytest -q

目录

src/mindagent/
  core/          状态机、ReActLoop、Runtime 和核心协议
  context/       上下文构造、图像注入和裁剪
  providers/     Provider、Router 和 Reasoner
  tools/         Tool、Registry、Executor 和内置工具
examples/        可运行示例
tests/           单元测试