AIForge

从零手写的企业级 AI 应用开发框架。用统一的协议（Model / Tool / Agent / Workflow / Knowledge / Memory）把大模型组装成可运行、可观测、可扩展的 AI 应用——核心仅依赖 pydantic，所有组件可审计、可替换、可离线测试。

Features

• Agent — 角色化 AI 工作者：name / model / prompt / tools / knowledge / config 组合
• Multi-Agent — Agent 即 Tool（AgentTool 递归合流），Supervisor / Router / Planner 协作编排
• Workflow — 线性与 DAG 流程编排：条件分支、并行执行、重试、人工审批（暂停/恢复）、YAML DSL
• RAG — Knowledge + Retriever 协议：关键词与向量检索（注入 embedder），结果自动注入 Prompt
• MCP — 真实 MCP 客户端：JSON-RPC 2.0 over stdio / HTTP(SSE)，外部系统工具一键接入
• Memory — Memory 协议 + 多后端：InMemory / SQLite / Redis / Postgres / Vector（语义检索），租户隔离
• Observability — 事件流 → Trace，每次执行可回放；流式 tokens/cost 指标；Prometheus 导出
• Evaluation — 数据集 / 评估器 / LLM Judge（四维评分）/ 基准对比
• Security — RBAC 权限（默认拒绝），门控工具、知识、Agent 与应用
• Runtime — 异步任务队列（InMemory / Redis）、执行状态机、检查点恢复、aiforge deploy 一键部署

Architecture

                Enterprise System
                       │
                FastAPI Adapter
                       │
                AIApplication   ← 统一调用门面（run / stream / trace）
                       │
                 Application Runtime（生命周期编排）
                       │
          ┌────────────┴────────────┐
          ↓                         ↓
       Agent                    Workflow
          │
      Strategy Layer（Router / ReAct / Reflection / Debate）
          │
       Executor Core（Tool Loop：Model ↔ Tool）
          │
   Model / Tool / Knowledge / Memory
          │
   Trace / Evaluation / Security / Observability

分层纪律：上层依赖协议、不依赖实现——BaseModel / BaseTool / Retriever / Memory / TaskQueue 都是协议，新增实现（供应商、存储、传输）不动内核。

Quick Start

pip install zsh-aiforge

from aiforge import AIApplication, Agent, OpenAIModel, Prompt
from aiforge.tools.base import BaseTool


class WeatherTool(BaseTool):
    name = "get_weather"
    description = "查询指定城市的天气"

    def _run(self, **kwargs):
        return f"{kwargs['city']}：晴，25℃"


agent = Agent(
    name="助手",
    model=OpenAIModel(model="gpt-4o-mini"),   # 没有 API Key 时用 MockModel() 即可离线运行
    prompt=Prompt(system="你是一个企业助手"),
    tools=[WeatherTool()],
)
app = AIApplication(agent=agent)

print(app.run("北京天气怎么样？").message.content)   # 同步
for chunk in app.stream("上海天气呢？"):             # 流式
    print(chunk.content, end="")
print(app.trace())                                  # 执行轨迹

PyPI 包名是 zsh-aiforge，代码导入名始终是 aiforge（from aiforge import ...），CLI 命令是 aiforge。

Example

完整可运行的示例见 examples/：

Core Concepts

• Agent — 决策者："这个任务该怎么做"。持有模型、工具、知识，由 Executor 驱动 Tool Loop。
• Workflow — 流程控制器："业务流程怎么走"。明确的 A → B → C → D，可条件分支、并行、重试、人工介入。
• Tool — 能力单元：本地工具、MCP 工具、外部 HTTP 工具、另一个 Agent（AgentTool）。
• Knowledge — 企业知识门面：问题 → 检索结果，自动注入 Prompt 的 {knowledge} 变量。
• Runtime — 生命周期编排：Context → Memory → Prompt → Middleware → Executor → Trace。
• Trace — 可观测性：事件流（run.started → model.response → tool.call/result → run.completed）构建为结构化执行轨迹，成败皆留痕。

Production Capabilities

• Security — Identity + RoleBasedPermissionChecker（默认拒绝），RBAC 门控四类资源
• Multi-Tenant — (tenant, user, session) 三元组隔离，SessionManager 会话生命周期
• Checkpoint — 执行现场快照 + app.resume() 失败恢复
• Persistent Trace — SQLiteTraceStore + TraceAPI（历史查询 / 时间线）
• Async — app.submit() 异步任务队列（InMemoryTaskQueue / RedisTaskQueue）、await app.arun() / async for chunk in app.astream()
• Distributed Runtime — 六态执行状态机、后台 Worker、aiforge deploy 生成 docker-compose / Dockerfile / API server / worker
• FastAPI Adapter — POST /run /stream /arun /astream /submit、GET /trace/{id} /runs /result/{id} /health /metrics
• Prompt Lifecycle — 版本管理（保存/回滚）、灰度实验（确定性分流）、成本路由

API Documentation

公共 API 参考见 docs/api.md；架构细节见 docs/architecture.md。

Development

pip install -e ".[dev]"

python -m pytest -q          # 935 tests passing
python -m mypy aiforge --ignore-missing-imports   # mypy clean

License

MIT