内容
# SimAgentPlg
[English](README.md) | [简体中文](README_zh-CN.md)
SimAgentPlg 0.2.3 是一个用于构建有状态的、兼容 OpenAI 的智能代理的轻量级框架，具备可组合的工具处理器、可选的 MCP 工具、本地技能路由和简单的基于角色的多代理工作流。
## 功能特点
- 有状态的 `BaseAgent`，具备对话记忆和显式的 `reset()` 方法
- 不可变的、必需的 `agent_id`，每个代理独有
- 通过 `.env` 文件或直接构造，兼容 OpenAI 的模型配置
- 可选的工具模式，具备显式的处理器注册
- 内置的 `BashHandler`，用于有限的 Bash 执行
- 内置的 `GitDiffHandler`，用于 Git 工作树检查
- 内置的 `FinishHandler`，用于显式的任务完成
- `MethodToolHandler`，用于小型自定义 Python 工具
- `AgentManager`，具备每代理序列化功能和跨代理并发能力
- 线性的 `AgentWorkflow`，用于计划者、执行者、审查者等类似角色
- 可选的 MCP 集成，通过 `McpToolHandler` 和 `McpServerManager`
- 可选的本地技能发现和路由，通过 `SkillManager`
Python 3.12 或更新版本是必需的。
## 安装
使用 `uv` 安装本地项目和依赖：
```bash
uv sync
```
## 配置
复制 `.env.example` 到 `.env`，然后填写模型凭证：
```env
MODEL_API_KEY=sk-xxxxxxxx
MODEL_URL=https://api.deepseek.com
CHAT_MODEL=deepseek-v4-flash
SKILL_MODEL=deepseek-v4-flash
LLM_TIMEOUT=60
LLM_TEMPERATURE=0.7
```
`ModelConfig.from_env()` 读取 `CHAT_MODEL`、`MODEL_API_KEY`、`MODEL_URL`、`LLM_TIMEOUT` 和 `LLM_TEMPERATURE`。
您也可以直接构造一个配置：
```python
from simagentplg import ModelConfig
config = ModelConfig(
model="deepseek-v4-flash",
api_key="sk-xxxxxxxx",
base_url="https://api.deepseek.com",
)
```
## 快速开始
### 普通聊天
工具执行默认情况下是禁用的。一个普通的代理在 `runtime()` 调用之间保持对话历史：
```python
from simagentplg import BaseAgent, ModelConfig
agent = BaseAgent(
config=ModelConfig.from_env(),
agent_id="tutor",
system_prompt="你是一个简洁的 Python 辅导员。",
)
first = await agent.runtime(task="记住我更喜欢 Python。")
second = await agent.runtime(task="你更喜欢哪种语言？")
agent.reset()
await agent.shutdown()
```
### 工具模式
设置 `enable_tools=True` 并显式传递处理器：
```python
import json
from simagentplg import (
BaseAgent,
BashHandler,
FinishHandler,
GitDiffHandler,
ModelConfig,
)
agent = BaseAgent(
config=ModelConfig.from_env(),
agent_id="developer",
system_prompt="使用可用的工具完成编码任务。",
handlers=[BashHandler(), GitDiffHandler(), FinishHandler()],
enable_tools=True,
)
result = await agent.runtime(task="创建打印 'hello' 的 hello.py。")
report = json.loads(result)
print(report["summary"])
await agent.shutdown()
```
启用工具的代理仅公开传递给 `BaseAgent` 的处理器：
```text
BaseAgent
-> BashHandler
-> bash_run
-> GitDiffHandler
-> run_gitdiff
-> FinishHandler
-> run_finish
-> MethodToolHandler 子类
-> McpToolHandler
```
在工具模式下，普通文本不会完成任务。模型必须调用完成工具，通常是 `run_finish`，或者自定义工具必须返回 `StepOutcome(..., should_exit=True)`。
当以下情况发生时，工具模式停止并报错：
- 在 `max_steps` 内未调用完成工具
- 连续三次请求相同的工具和参数
## 内置处理器
`BashHandler` 公开 `bash_run` 并执行有限的 Bash 命令。它有工作目录、超时和输出限制。
`GitDiffHandler` 公开 `run_gitdiff`。它检查当前的 Git 工作树，不会完成任务：
```json
{
"status": "success",
"mode": "status",
"command": "git status --short",
"output": "?? hello.py\n"
}
```
支持的模式有 `status`（`git status --short`）、`stat`（`git diff --stat`）和 `diff`（`git diff`）。
`FinishHandler` 公开 `run_finish`。它返回 JSON 结果并退出当前的 `runtime()`：
```json
{
"summary": "创建 hello.py"
}
```
## 工具中间件
`ToolMiddleware` 可以在执行前检查工具调用。框架不定义全局风险级别；应用程序可以编写自己的中间件或使用 `BashApprovalMiddleware` 在 `bash_run` 匹配风险命令模式时要求确认：
```python
from simagentplg import (
BaseAgent,
BashApprovalMiddleware,
BashHandler,
FinishHandler,
ModelConfig,
)
agent = BaseAgent(
ModelConfig.from_env(),
agent_id="coder",
handlers=[BashHandler(), FinishHandler()],
middlewares=[BashApprovalMiddleware()],
enable_tools=True,
)
```
## 自定义工具处理器
`MethodToolHandler` 将名为 `add` 的工具映射到名为 `do_add` 的异步方法：
```python
from collections.abc import Mapping
from typing import Any
from simagentplg import BaseAgent, MethodToolHandler, ModelConfig, StepOutcome
ADD_TOOL = {
"type": "function",
"function": {
"name": "add",
"description": "添加两个数字。",
"parameters": {
"type": "object",
"properties": {
"left": {"type": "number"},
"right": {"type": "number"},
},
"required": ["left", "right"],
},
},
}
class MathHandler(MethodToolHandler):
def __init__(self) -> None:
super().__init__((ADD_TOOL,))
async def do_add(self, arguments: Mapping[str, Any]) -> StepOutcome:
return StepOutcome(
{"value": arguments["left"] + arguments["right"]}
)
agent = BaseAgent(
config=ModelConfig.from_env(),
agent_id="calculator",
handlers=[MathHandler()],
enable_tools=True,
)
```
处理器启动时构建一个路由表。重复的工具名称被拒绝，而不是静默覆盖另一个处理器。
## 代理管理器
每个代理拥有自己的身份，因此注册不会重复 ID：
```python
from simagentplg import AgentManager, BaseAgent, ModelConfig
config = ModelConfig.from_env()
manager = AgentManager()
manager.register(
BaseAgent(
config=config,
agent_id="writer",
system_prompt="你编写简洁的发布说明。",
)
)
manager.register(
BaseAgent(
config=config,
agent_id="reviewer",
system_prompt="你审查软件变更的风险。",
)
)
results = await manager.run_many(
{
"writer": "为版本 0.2.3 编写发布说明。",
"reviewer": "审查兼容性风险。",
}
)
await manager.shutdown()
```
对同一代理的调用是序列化执行的，因为它们共享消息历史。对不同代理的调用可以并发执行。`run_many()` 以值形式返回失败结果，因此一个代理的失败不会取消其他代理。
`run_isolated(agent_id, task)` 重置并执行一个代理，同时保持相同的每代理锁。工作流使用它来避免角色或步骤之间的隐式历史泄漏。
## 基于角色的工作流
`AgentWorkflow` 以验证的线性管道执行代理角色：
```python
from simagentplg import (
AgentManager,
AgentWorkflow,
BaseAgent,
ModelConfig,
WorkflowStep,
)
config = ModelConfig.from_env()
manager = AgentManager()
manager.register(
BaseAgent(
config=config,
agent_id="planner",
system_prompt="创建简洁的实现计划。",
)
)
manager.register(
BaseAgent(
config=config,
agent_id="executor",
system_prompt="使用工具执行计划。",
enable_tools=True,
)
)
manager.register(
BaseAgent(
config=config,
agent_id="reviewer",
system_prompt="审查完成的工作的正确性和风险。",
)
)
workflow = AgentWorkflow(
manager,
[
WorkflowStep(
name="plan",
agent_id="planner",
prompt="计划此任务：\n{input}",
),
WorkflowStep(
name="execute",
agent_id="executor",
prompt=(
"原始任务：\n{original_task}\n\n"
"执行此计划：\n{input}"
),
),
WorkflowStep(
name="review",
agent_id="reviewer",
prompt="审查执行结果：\n{execute}",
),
],
)
result = await workflow.run("实现用户登录")
print(result.final_output)
await manager.shutdown()
```
工作流模板支持 `{input}`、`{original_task}` 和已完成命名步骤的输出，例如 `{plan}` 或 `{execute}`。未知变量和前向引用在创建工作流时被拒绝。版本 0.2.3 仅支持线性步骤。
## MCP 工具
MCP 是可选的，遵循相同的处理器约定：
```python
from simagentplg import BaseAgent, McpToolHandler, ModelConfig
agent = BaseAgent(
config=ModelConfig.from_env(),
agent_id="browser",
handlers=[McpToolHandler("example/mcp_config.json")],
enable_tools=True,
)
```
MCP 配置示例：
```json
{
"mcpServers": {
"playwright": {
"command": "npx",
"args": ["@playwright/mcp@latest", "--headless"]
}
}
}
```
`McpServerManager` 加载配置的服务，以服务名称前缀公开工具，并让一个失败的服务避免阻塞其他服务。
## 技能
技能是可选的提示扩展，独立于工具处理器：
```python
from pathlib import Path
from simagentplg import BaseAgent, ModelConfig
agent = BaseAgent(
config=ModelConfig.from_env(),
agent_id="skilled-agent",
skills_dir=Path("example/skills"),
enable_tools=True,
)
```
`SkillManager` 扫描每个包含 `SKILL.md` 的子目录。由 `SKILL_MODEL` 选择的路由模型从 YAML 前面内容中选择一个技能。选定的 `SKILL.md`、可选的 `template.md` 和可选的 `examples/sample.md` 被注入到代理上下文中。
```text
example/skills/
release_notes/
SKILL.md
template.md
examples/
sample.md
```
技能目前通过工具模式生命周期运行，因此设置 `enable_tools=True` 并使用 `run_finish` 完成。
## 示例
可在 [`example/`](example/README.md) 找到可运行的示例：
```bash
uv run python example/01_stateful_chat.py
uv run python example/02_custom_tool.py
uv run python example/03_multi_agent.py
uv run python example/04_mcp_tools.py
uv run python example/05_role_workflow.py
uv run python example/06_skill.py
uv run python example/07_bash_approval.py
```
## 测试
从仓库根目录运行测试套件：
```bash
uv run python -m unittest
```
当前测试涵盖代理、自定义处理器、工具中间件、完成行为、管理器锁定/并发、工作流和可导入的示例。
## 公共 API
```python
BaseAgent(
config: ModelConfig | None = None,
*,
agent_id: str,
system_prompt: str = REACT_LOOP_PROMPT,
handlers: Iterable[BaseHandler] | None = None,
middlewares: Iterable[MiddleWare] | None = None,
enable_tools: bool = False,
skills_dir: str | Path | None = None,
max_steps: int = 20,
client: Any | None = None,
)
await agent.runtime(*, task: str) -> str | None
agent.reset(history=None)
await agent.startup()
await agent.shutdown()
```
顶级包导出 `BaseAgent`、`ModelConfig`、`StepOutcome`、`AgentManager`、`workflow` 类型、`handler` 基类、`MethodToolHandler`、`BashHandler`、`GitDiffHandler`、`FinishHandler`、`McpToolHandler`、`handler` 错误、`McpServerManager`、`SkillManager` 和默认资源路径。
## 0.2.3 中的更改
- 添加了配套的 `FinishHandler` 和内置的 `run_finish` 工具
- 添加了配套的 `GitDiffHandler` 和内置的 `run_gitdiff` 工具
- 在工具模式下，需要显式完成工具
- 添加了对连续三次相同工具调用的保护
- 当工具模式耗尽 `max_steps` 时，引发明确的错误
- 保持 `BashHandler` 专注于 `bash_run`
## 许可证
MIT
