内容
# geny-executor
[](https://pypi.org/project/geny-executor/)
[](https://pypi.org/project/geny-executor/)
[](LICENSE)
[](https://github.com/CocoRoF/geny-executor/actions/workflows/ci.yml)
**一款工程化的智能代理管道库 — 21 个阶段，5 个 LLM 提供商，MCP 原生，全可 introspect。**
geny-executor 实现了一个 **21 阶段的管道**，具有 **双重抽象架构**（阶段槽 × 策略槽）。灵感来自 Claude Code 的智能代理循环和 Anthropic 的 harness 设计原则。没有 LangChain，没有 LangGraph，只是一个明确的、模块化的管道，每个步骤都是可观察的、可变的和可替换的。
[한국어 README](README_ko.md) · [架构](docs/architecture.md) · [提供商](docs/providers.md) · [错误代码](docs/error_codes.md) · [Claude Code CLI 主机](docs/claude_code_cli.md)
---
## Geny 生态系统
这些项目是为协同工作而构建的。**Geny** 是堆栈顶部的产品；下面每个项目都是一个可单独使用的构建块。**➡️ 标记当前位置。**
| 项目 | 描述 | 在堆栈中的角色 |
|---|---|---|
| [**Geny**](https://github.com/CocoRoF/Geny) | 多智能体 VTuber + 自主工作器平台 | 产品 — 使用以下所有项目 |
| ➡️ [**geny-executor**](https://github.com/CocoRoF/geny-executor) | 21 阶段、基于清单的智能代理管道 · PyPI · Apache-2.0 | 引擎 — 所有项目运行在此之上 |
| [**GAPT**](https://github.com/CocoRoF/geny-adapted-project-toolkit) | 自托管 AI DevOps 平台 — 沙盒 · 编辑 · 构建 · 部署 | 智能体安全触及真实仓库的地方 |
| [**geny-avatar**](https://github.com/CocoRoF/geny-avatar) | 2D live-avatar 编辑器，带有 AI 纹理生成 | Geny 的面部制作地点 |
<details>
<summary>它们如何协同工作</summary>
```
Geny — 产品（使用以下所有内容）
│
┌─────────────┼──────────────┐
智能体引擎 avatars 沙盒 + 部署
│ │ │
▼ ▼ ▼
geny-executor geny-avatar GAPT
（引擎） （avatar 编辑器） （AI DevOps 平台）
```
</details>
---
<!-- 📸 IMAGE NEEDED: hero banner — the 21-stage pipeline as a clean horizontal flow graphic -->
> 📸 **需要图像** — _英雄横幅：21 阶段管道呈现为精致的流程图。_
---
## 为什么选择 geny-executor？
| 问题 | geny-executor 的答案 |
|---|---|
| 框架隐藏太多抽象内容 | 21 个阶段中的每一个都是明确的、可检查的和可单独替换的。 |
| 难以在不重写所有内容的情况下自定义一个部分 | **双重抽象**：交换整个阶段 *或* 交换阶段内的策略。基于清单的配置 = 工件。 |
| LLM 提供商之间的供应商锁定 | 一个合同，五个提供商（`anthropic` / `openai` / `google` / `vllm` / `claude_code_cli`）。通过编辑一个配置字段进行切换。 |
| 智能体循环是不透明的黑盒 | 事件总线 + 稳定的结构化错误代码（[`exec.cli.auth_failed`，…](docs/error_codes.md)）— 每个故障都在日志 / Sentry / i18n 层中清晰分组。 |
| MCP 集成是一个次要考虑因素 | 一流的。主机附加 MCP 服务器 + 用于 CLI 后端的每个会话 MCP 包装（例如 Claude Code CLI）出厂即用。 |
| 成本跟踪是一个事后想法 | 集成到第 7 阶段（Token）。每次调用的成本，每个会话的分类账，预算保护。 |
---
## 架构概览
### 21 阶段管道
```
阶段 A — 设置（每次循环一次）
1：输入 → 2：上下文 → 3：系统 → 4：守卫 → 5：缓存
阶段 B — 生成 + 分发（循环）
6：API → 7：Token → 8：思考 → 9：解析
→ 10：工具 → 11：工具审查 → 12：智能体 → 13：任务注册表
→ 14：评估 → 15：HITL → 16：循环
阶段 C — 表面（一次）
17：发出 → 18：内存 → 19：总结 → 20：持久化 → 21：产生
```
完整阶段列表及策略选项参见 [`docs/architecture.md`](docs/architecture.md)。
### 双重抽象 — 两个层次的交换
```
┌─ 层次 1：阶段抽象 ─────────────────────────┐
│ 交换管道中整个阶段模块的进出口。 │
│ │
│ ┌─ 层次 2：策略抽象 ─────────────────┐ │
│ │ 在阶段内交换内部逻辑。 │ │
│ │ │ │
│ │ 上下文阶段可以使用： │ │
│ │ → 简单加载 （默认） │ │
│ │ → 进步披露 │ │
│ │ → 向量搜索 │ │
│ │ → 自定义策略 │ │
│ └──────────────────────────────────────────────────┘ │
└────────────────────────────────────────────────────────┘
```
- **阶段抽象** — 替换整个阶段（例如，替换私有提供商的自定义 `APIStage`）。
- **策略抽象** — 在阶段内更改行为（例如，从 `SimpleLoad` 切换到 `VectorSearch`）而无需触及周围的管道。
---
## 安装
```bash
pip install geny-executor
```
可选额外功能：
```bash
pip install geny-executor[memory] # 用于向量检索的 numpy
pip install geny-executor[all] # 所有内容
pip install geny-executor[dev] # 开发/测试工具
```
**要求**：Python 3.11+。至少一个提供商的凭证（Anthropic API 密钥、OpenAI API 密钥，...）或本地 CLI 二进制文件（`claude` 用于 `claude_code_cli`）。
---
## 快速入门
### 最小管道
```python
import asyncio
from geny_executor import PipelinePresets
async def main():
pipeline = PipelinePresets.minimal(api_key="sk-ant-...")
result = await pipeline.run("法国的首都是什么？")
print(result.text)
asyncio.run(main())
```
### 聊天管道（历史 + 系统提示 + 可选工具）
```python
from geny_executor import PipelinePresets
pipeline = PipelinePresets.chat(
api_key="sk-ant-...",
system_prompt="你是一个有帮助的编码助手。",
)
result = await pipeline.run("解释 Python 装饰器")
print(result.text)
print(f"成本：${result.total_cost_usd:.4f}")
```
### 完整智能体（所有 21 个阶段 — 工具、评估、内存、循环控制）
```python
from geny_executor import PipelinePresets
from geny_executor.tools import ToolRegistry, Tool, ToolResult, ToolContext
class SearchTool(Tool):
@property
def name(self) -> str: return "search"
@property
def description(self) -> str: return "在网络上搜索信息"
@property
def input_schema(self) -> dict:
return {
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"],
}
async def execute(self, input, context):
return ToolResult(content=f"{input['query']} 的结果")
registry = ToolRegistry()
registry.register(SearchTool())
pipeline = PipelinePresets.agent(
api_key="sk-ant-...",
system_prompt="你是一个研究助手。使用工具查找答案。",
tools=registry,
max_turns=20,
)
result = await pipeline.run("查找最新的 Python 发布版本")
```
### 使用构建器自定义管道
```python
from geny_executor import PipelineBuilder
pipeline = (
PipelineBuilder("my-agent", api_key="sk-ant-...")
.with_model(model="claude-sonnet-4-6", max_tokens=4096)
.with_system(prompt="你是一个简洁的助手。")
.with_context()
.with_guard(cost_budget_usd=1.0, max_iterations=30)
.with_cache(strategy="aggressive")
.with_tools(registry=my_registry)
.with_think(enabled=True, budget_tokens=10000)
.with_evaluate()
.with_loop(max_turns=30)
.with_memory()
.build()
)
result = await pipeline.run("复杂的分步任务")
```
### 基于清单的管道（推荐用于主机）
```python
from geny_executor import Pipeline, CredentialBundle, ProviderCredentials, EnvironmentManifest
manifest = EnvironmentManifest.load("./envs/my_env.json")
credentials = CredentialBundle(by_provider={
"anthropic": ProviderCredentials(api_key="sk-ant-..."),
})
pipeline = await Pipeline.from_manifest_async(manifest, credentials=credentials)
result = await pipeline.run("你好！")
```
参见 [`docs/manifest.md`](docs/manifest.md) 了解完整架构。
---
## 五个 LLM 提供商，一个合同
| 提供商 | 备注 |
|---|---|
| `anthropic` | 克劳德家族。完整的流式传输，原生 `tool_use`，思考块。 |
| `openai` | GPT-4.1 / o 系列。流式传输，工具，JSON 模式结构化输出。 |
| `google` | Gemini 3.x / 2.5。流式传输，工具，思考块。 |
| `vllm` | 任意模型在本地 vLLM 端点。OpenAI 兼容。工具通过 `configure_capabilities()` 按需启用。 |
| `claude_code_cli` | 子进程驱动的 Claude Code CLI。**主机附加每个会话的 MCP 桥** 以向生成的 CLI 的 LLM 公开其自己的工具注册表。参见 [`docs/claude_code_cli.md`](docs/claude_code_cli.md)。 |
会话通过 `stages[6].config["provider"]` 在清单中选择提供商。凭证通过单个 `CredentialBundle` 通道流动 — 参见 [`docs/providers.md`](docs/providers.md)。
---
## 错误代码（2.1.0+）
每个执行器异常都带有稳定的 `exec.<component>.<reason>` 代码：
```python
from geny_executor import APIError, ExecutorErrorCode, ErrorCategory
try:
result = await pipeline.run("...")
except APIError as e:
if e.code is ExecutorErrorCode.EXEC_CLI_AUTH_FAILED:
print("请重新登录到 Claude Code CLI.")
elif e.category.is_recoverable:
print(f"可恢复的故障（{e.code.value}）；重试中。")
```
结构化事件有效载荷也包含代码：
```json
{
"type": "pipeline.error",
"data": {
"error": "Claude Code CLI 未认证 …",
"code": "exec.cli.auth_failed",
"exception_type": "geny_executor.core.errors.APIError"
}
}
```
代码在版本间 **稳定** — 参见 [`docs/error_codes.md`](docs/error_codes.md) 了解完整表格、可恢复性和如何添加新代码。
---
## 会话
跨多个交互的持久状态：
```python
from geny_executor import PipelinePresets
from geny_executor.session import SessionManager
manager = SessionManager()
pipeline = PipelinePresets.chat(api_key="sk-ant-...")
session = manager.create(pipeline)
await session.run("我的名字是 Alice")
result = await session.run("我的名字是什么？")
for info in manager.list_sessions():
print(f"{info.session_id}: {info.message_count} 条消息，${info.total_cost_usd:.4f}")
```
---
## 事件系统 + 可观察性
```python
@pipeline.on("stage.enter")
async def _(event):
print(f"→ {event.stage}")
@pipeline.on("pipeline.error")
async def _(event):
print(f"❌ {event.data['code']}: {event.data['error']}")
@pipeline.on("*")
async def _(event):
pass # 消息队列
```
流式传输：
```python
async for event in pipeline.run_stream("逐步解决"):
if event.type == "stage.enter":
print(f"阶段：{event.stage}")
elif event.type == "pipeline.complete":
print(f"最终结果：{event.data['result'].text}")
```
---
## 工具 + MCP
```python
from geny_executor.tools import Tool, ToolResult, ToolContext, ToolRegistry
class Calculator(Tool):
@property
def name(self): return "calculator"
@property
def description(self): return "执行算术运算。"
@property
def input_schema(self):
return {"type": "object", "properties": {"expression": {"type": "string"}}, "required": ["expression"]}
async def execute(self, input, context):
return ToolResult(content=str(eval(input["expression"]))) # 使用安全评估器！
registry = ToolRegistry()
registry.register(Calculator())
```
连接主机附加的 MCP 服务器：
```python
from geny_executor.tools.mcp import MCPManager
mcp = MCPManager()
await mcp.connect("filesystem", command="npx", args=["-y", "@anthropic/mcp-filesystem"])
for tool in mcp.list_tools():
registry.register(tool)
```
有关 **CLI 端** MCP 包装（您的工具注册表暴露到生成的 Claude Code CLI 的 LLM 中），请参见 [`docs/claude_code_cli.md`](docs/claude_code_cli.md)。
---
## 管道预设
| 预设 | 活动阶段 | 用例 |
|---|---|---|
| `PipelinePresets.minimal()` | 输入 → API → 解析 → 产生 | 快速问答，烟雾测试 |
| `PipelinePresets.chat()` | + 上下文，系统，守卫，缓存，令牌，工具，循环，内存 | 会话式聊天机器人 |
| `PipelinePresets.agent()` | 所有 21 个阶段激活 | 具有工具，评估，内存，总结，持久化的自主智能体 |
| `PipelinePresets.evaluator()` | 输入 → 系统 → API → 解析 → 评估 → 产生 | 生成器/评估器质量通道 |
| `PipelinePresets.geny_vtuber()` | 所有 21 个阶段 + VTuber/TTS 发射器 | Geny VTuber 套件的参考复制品 |
---
## 自定义阶段 + 策略
```python
from geny_executor.core.stage import Strategy
class MyContextStrategy(Strategy):
name = "my_context"
description = "使用 RAG 的自定义上下文加载"
def configure(self, config: dict) -> None:
self.top_k = config.get("top_k", 5)
async def load(self, state):
... # 您的 RAG 检索
```
```python
from geny_executor.core.stage import Stage
from geny_executor.core.state import PipelineState
class LoggingStage(Stage[dict, dict]):
name = "logging"
order = 7 # 在 API 之后，在思考之前
category = "execution"
async def execute(self, input, state: PipelineState):
print(f"[{state.iteration}] 收到 API 响应")
return input
pipeline.register_stage(LoggingStage())
```
---
## 项目结构
```
geny-executor/
├── src/geny_executor/
│ ├── __init__.py # 公共 API 表面
│ ├── py.typed # PEP 561 类型标记
│ ├── core/ # 管道引擎，错误，清单，突变，快照
│ ├── stages/ # 21 个管道阶段（s01–s21）
│ ├── llm_client/ # 5 个提供商 + ClientRegistry + CredentialBundle + CLI 运行时
│ ├── tools/ # 工具 ABC，注册表，路由器，MCP 集成
│ ├── hooks/ # PRE/POST 工具使用生命周期钩子
│ ├── memory/ # 内存 v2 检索，保险库映射，矢量存储
│ ├── skills/ # 技能提供商 + 技能加载
│ ├── subagents/ # 第 12 阶段子智能体编排
│ ├── permission/ # 每个工具的 ACL 由 RegistryRouter 评估
│ ├── channels/ # 输出通道适配器（文本，回调，TTS，...）
│ ├── cron/ # 调度触发器支持
│ ├── events/ # 事件总线发布/订阅
│ ├── history/ # 会话历史原始数据
│ ├── telemetry/ # 事件/指标导出器
│ └── session/ # 会话管理器 + 新鲜度检查
├── docs/ # 架构，提供商，清单，错误代码，MCP，钩子
├── tests/ # 3100+ 个单元测试，一致性测试，契约测试，集成测试
├── pyproject.toml # 包配置（Hatch）
└── LICENSE # Apache-2.0
```
---
## 开发
```bash
git clone https://github.com/CocoRoF/geny-executor.git
cd geny-executor
pip install -e ".[dev]"
pytest # 完整测试套件（~30 秒，3100+ 测试）
pytest tests/contract/test_error_codes_stability.py # 错误代码稳定性检查
pytest --cov=geny_executor --cov-report=term-missing # 覆盖率
ruff check src/ tests/
ruff format src/ tests/
```
---
## 版本控制
| 版本 | 亮点 |
|---|---|
| **2.1.0** | `ExecutorErrorCode` 分类法 + 结构化 `pipeline.error` / `stage.error` / `api.retry` 载荷。 `docs/error_codes.md`。 |
| **2.0.6** | 删除 `copilot_cli` 提供商（纯文本，无主机工具往返）。上游 Geny 的 claude_code_cli 兼容性补丁（`--verbose` 注入，`--bare` 剥离，自动删除 `--tools ""`，`tool_use` 从 finalize 中剥离）。 |
| **2.0.5** | `APIRequest.mcp_config` 每请求重写 + 自动发出 `--strict-mcp-config`。主机 MCP 包装的基础支持。 |
| **2.0.0** | 提供商抽象（`ClientRegistry`，`CredentialBundle`）。第 6 阶段提供商的清单单一事实来源。 |
| **1.x** | 原始的 16 阶段管道；仅 Anthropic。 |
参见 [CHANGELOG](https://github.com/CocoRoF/geny-executor/releases) 了解完整的历史。
---
## 许可证
[Apache License 2.0](LICENSE)。版权所有 2026 CocoRoF — 参见 [NOTICE](NOTICE)。
---
## 相关项目
**Geny 生态系统**（与此引擎一起构建的 sibling 项目）→ 参见 [The Geny ecosystem](#the-geny-ecosystem) 上方：
[Geny](https://github.com/CocoRoF/Geny) · [GAPT](https://github.com/CocoRoF/geny-adapted-project-toolkit) · [geny-avatar](https://github.com/CocoRoF/geny-avatar)
**已构建并与以下项目互操作：**
- [Anthropic SDK](https://github.com/anthropics/anthropic-sdk-python)
- [OpenAI SDK](https://github.com/openai/openai-python)
- [Google GenAI SDK](https://github.com/googleapis/python-genai)
- [vLLM](https://github.com/vllm-project/vllm)
- [Claude Code CLI](https://docs.anthropic.com/claude/code/) — geny-executor 通过 `claude_code_cli` 提供商托管它
- [MCP](https://modelcontextprotocol.io/) — 模型上下文协议；主机附加服务器和会话 CLI 包装均为一流公民
MCP 配置
以下是此 MCP Server 的配置信息，可直接复制到 Cursor 或其他 MCP 客户端中使用。
mcp.json
连接信息
猜你喜欢
暂无推荐服务器