AIForge

一个从零手写的企业级 AI 应用开发框架（AI 版 Spring Boot 的实现范本）。 Model、Tool、Agent、Knowledge、Workflow、MCP、Memory、Prompt、Middleware、Trace、Runtime 全部自研, 用 AIApplication 把零件组装成可运行、可观测、可扩展的 AI 应用。

双重定位：

• 学习者：想理解 AI 框架原理（协议怎么设计、Tool Loop 怎么转、事件流怎么变成 Trace）—— AIForge 是按 28 个阶段逐步演进出来的完整范本：核心零依赖（仅 pydantic）、全链路可审计、935 个测试可离线跑
• 轻量使用者：需要一个透明、可掌控的 AI 应用底座——不想要 LangChain 的重量与黑盒时， 用它组装 Agent / Workflow / RAG / MCP 应用

与 LangChain 的定位差异：互补而非替代 —— LangChain 生态全、抽象重、可快速堆功能； AIForge 轻、透明、可讲清全链路，适合理解原理与轻量落地。详见「为什么自研」与「已知边界」。

企业不用从零开发每一个 AI 应用,而是在 AIForge 上组装 AI 应用。

当前状态

19 个阶段(01 Core → 19 Architecture)全部完成,阶段 20(开发者体验)三轮全部完成, 阶段 21(生产运行增强)、阶段 22(企业运行平台层)、阶段 23(高级工作流引擎)全部完成, 阶段 24(企业集成层)、阶段 25(Agent 协作框架)、阶段 26(Agent 智能策略层)、 阶段 27(企业 AI 应用平台)、阶段 28(生产级运行基础设施)全部完成: 935 个测试全部通过、mypy 类型检查零错误、pip install / CLI / YAML 配置 / API 文档之外, 现在具备：执行保护(max_iterations / timeout / stop_reason)、工具可靠性(retry / timeout / 标准错误)、 运行指标(MetricsMiddleware / token / cost)、流式观测(chunk → event → trace)、自动生命周期(with / lifespan / FastAPI 对接), 企业运行平台能力——安全权限(Identity / PermissionChecker / Tool+Knowledge 权限)、 多租户会话(Session / SessionManager / Memory 三元组隔离)、执行检查点(CheckpointStore / resume 恢复)、 持久化 Trace(SQLiteTraceStore / 历史查询 / 失败分析)、评估体系(Dataset / Evaluator / Runner / trace_id 关联), 企业流程执行引擎——DAG 工作流图(WorkflowGraph / 拓扑调度)、条件分支(ConditionNode)、 并行执行(ParallelNode / asyncio gather)、流程级重试(RetryNode)、人工介入(HumanApprovalNode / 暂停+Checkpoint+Resume)、 工作流 Trace 树(WorkflowTraceStep / 嵌套 AgentTrace)、Workflow DSL(YAML 配置化工作流), 企业集成层——FastAPI Adapter(POST /run /stream + GET /trace /health /metrics,SSE 流式)、 持久化记忆(SQLiteMemory / 重启不丢 / 租户隔离)、真实 MCP transport(stdio JSON-RPC / HTTP SSE)、 外部工具网关(ExternalTool / 企业 HTTP API 统一成 BaseTool)、部署配置(YAML → ApplicationDefinition)、 健康与监控(HealthChecker 组件探活 / Prometheus 指标导出), 以及 Agent 协作框架——能力声明(AgentCapability / 注册+检索)、Agent 目录(AgentRegistry / 按能力找 Agent)、 协作编排(SupervisorAgent / 理解目标→选择→分配→汇总)、任务规划(PlannerAgent / 模型规划 JSON 步骤)、 共享上下文(CollaborationContext / 前序产出注入 Prompt)、协作 Trace(CollaborationTrace / 嵌套 ExecutionTrace)、 协作权限(AgentPermissionPolicy / Identity + Permission 控制「谁可以调用谁」), 以及 Agent 智能策略层——动态路由(AgentRouter / 模型排序选 Agent + fallback)、 推理状态(ReactStrategy / Thought-Action-Observation)、自我修正(ReflectionStrategy / 起草→检查→修订)、 多 Agent 辩论(DebateStrategy / 分析师立场→Judge 裁决)、三层记忆(ScopedMemory / Global→Agent→Session)、 向量检索(EmbeddingRetriever / embedding 相似度 + metadata 过滤)、 流式协作(CollaborationEvent / stream_run 事件流), 以及企业 AI 应用平台层——Prompt 版本管理(PromptVersion / PromptRegistry / 保存+版本+回滚)、 灰度实验(PromptExperiment / 确定性分流 / 统计对比 / 赢家发布)、 LLM Judge(LLMJudge / correctness+relevance+hallucination+citation 四维 0-10 评分)、 Agent 基准(BenchmarkSuite / 同一批任务对比 accuracy+latency+cost 排名)、 成本优化(CostController / 简单问题→便宜模型、复杂问题→推理模型的路由)、 观测面板(TraceAPI / 执行时间线 user_input→model→tool→final + FastAPI /runs 端点)、 应用治理(ApplicationPermissionPolicy / 谁可以用哪个 AI 应用、入口拦截), 以及生产级运行基础设施——分布式运行时(Task / TaskQueue / ExecutionScheduler / RuntimeWorker, 长任务不阻塞请求、worker 后台消费、失败隔离)、持久化执行引擎(ExecutionEngine 六态状态机 created/running/waiting/paused/completed/failed,非法转换拦截)、队列系统(app.submit() → {execution_id, status}, app.result(id) 取回)、异步原生运行时(await app.arun() / async for chunk in app.astream()), 分布式记忆(RedisMemory / PostgresMemory / VectorMemory,多实例共享 + TTL + 热数据 + 语义检索)、 生产部署(aiforge deploy 一键生成 docker-compose / Dockerfile / api_server / worker / deploy.yaml)。 定位升级:Enterprise AI Application Runtime Platform——不只是管理 AI 应用, 而是真正承载企业生产环境中的 AI 应用运行。

• 完整进度与交接信息:HANDOFF.md
• 4 张架构图:docs/architecture.md
• 28 阶段白皮书:docs/whitepaper.md
• 公共 API 参考:docs/api.md
• PyPI 发布指南:docs/PUBLISH.md
• 阶段 21 验收脚本:examples/test_phase21.py
• 阶段 22 验收脚本:examples/test_phase22.py
• 阶段 23 验收脚本:examples/test_phase23.py
• 阶段 24 验收脚本:examples/test_phase24.py
• 阶段 25 验收脚本:examples/test_phase25.py
• 阶段 26 验收脚本:examples/test_phase26.py
• 阶段 27 验收脚本:examples/test_phase27.py
• 阶段 28 验收脚本:examples/test_phase28.py

                Enterprise System
                           │
                    FastAPI Adapter
                           │
                    AIApplication  ← 应用治理闸门(谁可以用哪个应用)
                           │
                Application Runtime
                           │
          ┌────────────────┴────────────────┐
          ↓                                 ↓
       Agent                            Workflow
          │
      Strategy Layer (策略层)
          │
   Router / ReAct / Reflection / Debate
          │
       Executor Core (稳定核心)
          │
 Model / Tool / Knowledge / Memory
          │
 Trace / Evaluation / Governance
          │
 Prompt 版本+灰度实验 │ LLM Judge │ Benchmark │ Cost 路由 │ Trace API

    ┌─────────────────────────────────────────────┐
    │  Production Infrastructure (28)              │
    │  TaskQueue → Scheduler → Worker              │
    │  ExecutionEngine (状态机)                    │
    │  submit/result │ arun/astream │ deploy       │
    │  Redis / Postgres / Vector Memory            │
    └─────────────────────────────────────────────┘

5 分钟快速上手

1. 安装

# 方式一：从 PyPI 安装（已发布）
pip install zsh-aiforge
# 可选依赖按需安装：pip install "zsh-aiforge[openai,yaml,http]"

# 方式二：源码安装（开发 / 贡献）
git clone https://gitee.com/Zssssssh/AIForge.git
cd AIForge
pip install -e .

注意：PyPI 包名是 zsh-aiforge，代码里的导入名始终是 aiforge（from aiforge import ...），CLI 命令是 aiforge——包名与导入名解耦。

2. 配置 API Key（可选，不配也能用 MockModel 跑通）

# Windows PowerShell
$env:OPENAI_API_KEY = "sk-your-key-here"

# macOS / Linux
export OPENAI_API_KEY=sk-your-key-here

3. 创建你的第一个 AI 应用

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


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

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


# 用真实 OpenAI 模型
agent = Agent(
    name="企业助手",
    model=OpenAIModel(model="gpt-4o-mini"),
    prompt=Prompt(system="你是一个企业助手"),
    tools=[WeatherTool()],
)
app = AIApplication(agent=agent)

# 同步调用
response = app.run("北京天气怎么样？", session_id="user-001")
print(response.message.content)

# 流式调用
for chunk in app.stream("上海天气怎么样？", session_id="user-001"):
    print(chunk.content, end="")

# 查看执行追踪
print(app.trace())

4. 暂时没有 API Key？用 MockModel 快速体验

from aiforge.models.mock import MockModel

agent = Agent(
    name="测试助手",
    model=MockModel(),
    prompt=Prompt(system="你是一个测试助手"),
)
app = AIApplication(agent=agent)
print(app.run("你好").message.content)
# → MockModel 收到请求：你好

5. 用 CLI 一键生成项目（含配置 / 测试 / README）

aiforge init my-app      # 生成 app.py + config.yaml + test_app.py + README.md + .gitignore
cd my-app
python app.py            # 无 API Key 自动降级 MockModel，有 Key 走真实 OpenAI
python -m pytest test_app.py -v   # 生成的测试直接可用

生成的项目通过 config.yaml 配置（可被 manifest_from_yaml_file() 加载）， 支持 aiforge run <file> 运行、aiforge test 跑测试套件、aiforge --version 看版本。

快速开始（详细）

from aiforge import Agent, AIApplication, Prompt
from aiforge.models.mock import MockModel
from aiforge.tools.base import BaseTool


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

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


# 组装一个 Agent 应用
agent = Agent(
    name="企业助手",
    model=MockModel(),
    prompt=Prompt(system="你是一个企业助手"),
    tools=[WeatherTool()],
)
app = AIApplication(agent=agent)

# 统一 Invocation:字符串进,ModelResponse 出
response = app.run("北京天气怎么样？", session_id="demo-user")
print(response.message.content)

# 追踪:execution_id / status / duration / steps / error
print(app.trace())

# 流式:逐个 chunk 输出
for chunk in app.stream("上海天气怎么样？", session_id="demo-user"):
    print(chunk.content, end="")

更完整的用法(Workflow 应用 / Knowledge / MCP / Multi-Agent / Application Registry): 看 examples/ 下的 7 个验收脚本,最完整的是 application_platform_app.py。

学习路线（从原理到实践，按顺序读）

1. 跑通最小闭环：examples/basic_app.py（工具循环）→ examples/knowledge_app.py（RAG）→ examples/test_memory.py（多轮记忆）
2. 读核心链路：docs/architecture.md 图②「一次请求的生命线」→ 对照源码走一遍 app.run()： aiforge/application/app.py（门面）→ runtime/runtime.py（生命周期）→ core/executor.py（Tool Loop）→ core/trace.py（事件流 → Trace）
3. 理解协议设计：docs/whitepaper.md（28 阶段演进史——每阶段解决什么问题、留下什么协议、钉死什么边界）
4. 看完整应用示例：examples/medical_bot.py（RAG + Agent + 记忆组装）→ examples/multi_agent_app.py（Agent 调用 Agent，递归合流）
5. 面试/述职讲稿：docs/interview-walkthrough.md（app.run() 全链路讲述脚本 + 高频追问应答）
6. 动手扩展：实现一个新 Retriever / Memory / TaskQueue 实现，亲身体验「上层依赖协议、不依赖实现」

测试

python -m pytest -q          # 935 passed
python -m mypy aiforge --ignore-missing-imports   # 类型检查：零错误
python examples\application_platform_app.py   # 平台层验收
python examples\test_phase21.py   # 阶段 21 验收（执行保护/工具可靠性/指标/流式观测/生命周期）
python examples\test_phase22.py   # 阶段 22 验收（权限/多租户/检查点/持久化Trace/评估）
python examples\test_phase23.py   # 阶段 23 验收（DAG/条件分支/并行/重试/人工介入/Trace树/DSL）
python -X utf8 examples\test_phase24.py   # 阶段 24 验收（HTTP接入/持久化记忆/MCP transport/外部工具/部署/健康）
python -X utf8 examples\test_phase25.py   # 阶段 25 验收（能力/注册表/Supervisor/Planner/共享上下文/协作Trace/权限）
python -X utf8 examples\test_phase26.py   # 阶段 26 验收（Router/ReAct/Reflection/Debate/ScopedMemory/VectorRetriever/事件流）
python -X utf8 examples\test_phase27.py   # 阶段 27 验收（Prompt版本+灰度/LLMJudge/Benchmark/成本路由/TraceAPI/治理）
python -X utf8 examples\test_phase28.py   # 阶段 28 验收（分布式运行时/状态机/队列/异步原生/分布式记忆/deploy）

已知缺口与边界

✅ 已发布到 PyPI：pip install zsh-aiforge（当前 0.1.x，见 pypi.org/project/zsh-aiforge，发布步骤见 docs/PUBLISH.md）

功能缺口（正在演进）：

• Tool 硬超时需显式开启:ToolConfig(threaded=True, timeout=...) 走线程池硬超时、超时立即返回;默认仍为同步 + 事后检测(向后兼容)
• 流式 metrics 已覆盖:tokens 统计依赖 Provider 在流尾产出 usage 尾包(DeepSeek Provider 已支持;自定义 Provider 在 ModelStreamChunk 携带 usage 即自动生效)
• 任务队列:进程内 InMemoryTaskQueue 与跨进程 RedisTaskQueue 已就绪;Kafka / RabbitMQ 尚未实现(实现同一 TaskQueue 协议即可接入)
• 统一异常基类 AIForgeException 已落地(模型层已接入),其余各层异常逐步接入中

定位边界（如实声明）：

• 生态:不追求第三方集成生态——刻意保持轻量、可审计;需要生态时与 LangChain 等互补使用,而非替代
• 维护:个人维护、Alpha 阶段;适合学习与轻量业务起步,生产级企业请自行评估后使用
• 生产件清单:任务队列(InMemory/Redis)与检查点(内存)已就绪;Kafka、原生异步执行、外部向量库接入尚未实现——协议均已冻结,按协议补实现即可,不动内核

为什么自研:技术定位与取舍(面试视角)

AIForge 从零手写 Model / Tool / Agent / Workflow / RAG / MCP / Memory 协议, 而不是直接依赖 LangChain / LangGraph,是一组有意的设计决策:

这套取舍的收益:

• 可讲清全链路:app.run() 一次请求 = Runtime 编排生命周期 → Executor Tool Loop → 事件流 → Trace,每一环都有对应代码;
• 可离线验收:MockModel / MockLLM 让 935 个测试不依赖真实 API,全部可离线跑;
• 边界可审计:Application 只组装不执行、Agent 是决策者、Workflow 是流程控制器——职责钉死,扩展只加协议实现。

核心设计原则

• 上层依赖协议,不依赖实现:Agent 不认识 MCP / Vector DB / 其他 Agent 的内部结构
• 边界钉死:Application 是组装层不执行;Agent 是决策者,Workflow 是流程控制器;Memory 是档案室
• 失败不静默、可取证:工具失败不中断执行;异常挂执行现场原样上抛;成功与失败都保留 Trace
• 不创造无职责抽象:先跑起来,证明协议对,再加实现