mindagent 0.1.0

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

MindAgent

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

当前版本聚焦单 Agent、单 Action、串行 Tool Calling 的核心闭环。

当前能力

• 状态机控制 Agent 生命周期
• ReAct：Think → Validate → Execute → Observe → Update
• OpenAI-compatible Provider
• Provider 能力路由
• Tool 注册、JSON Schema 和参数校验
• 普通工具与流式工具
• ContextManager 文本、图像和 Tool Observation 管理
• 上下文长度估算与旧轮次裁剪
• 事件、heartbeat、超时、取消和人工审批
• JSONL 执行轨迹与离线回放
• 内置时间、计算器、上下文查询、进程内记忆和图像理解工具

MCP、Skill、长期记忆和并行 ToolCall 尚未实现。

架构

AgentRuntime
  ├── ReActLoop
  │     ├── ProviderReasoner
  │     │     └── ProviderRouter → LLM/VLM Provider
  │     ├── PolicyEngine
  │     ├── ToolExecutor → ToolRegistry → Tool
  │     └── ContextManager
  ├── EventHandler
  ├── Heartbeat
  └── Timeout / Cancel

核心边界：

• core 只负责编排和协议，不依赖具体模型或工具。
• providers 将不同模型接口统一为 ProviderResponse。
• tools 不感知 Agent 推理，只接收参数和 ToolContext。
• 所有外部动作通过 ActionRequest 表达。
• 每次 ToolCall 最终只生成一个 Observation。

环境要求

• Python 3.12+
• 项目虚拟环境 .venv
• OpenAI-compatible API

安装依赖：

uv pip install --python .venv/bin/python -r requirements.txt

也可以使用已激活虚拟环境中的 pip：

pip install -r requirements.txt

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。

内置工具

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

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

• time_now：获取指定 IANA 时区的当前时间
• calculator：基于受限 AST 计算算术表达式
• context_query：读取当前 run 的 metadata 或 artifacts
• memory：按 run 隔离的进程内 key/value 存储
• 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。

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
• 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。

测试

PYTHONPATH=src .venv/bin/python -m unittest discover -s tests -v

目录

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