DEV Community: 李春川

Monorepo 8 端同步开发:Turborepo + pnpm Workspace 实战

李春川 — Tue, 28 Jul 2026 05:38:41 +0000

Monorepo 8 端同步开发:Turborepo + pnpm Workspace 实战

做一个 AI 应用,产品经理说要上 Web、小程序、桌面端、浏览器插件、CLI、移动端 6 个端,加上 API 和 AI 服务共 8 个工程。是开 8 个仓库各干各的,还是用一个 monorepo 统管?IHUI-AI 选了后者。本文是我们在 8 端 monorepo 上用 Turborepo + pnpm Workspace 的工程总结,包含构建编排、依赖管理、跨端类型安全、CI 优化的完整实践。

一、为什么选 monorepo

1.1 polyrepo 的痛点

8 个端开 8 个仓库,会遇到:

类型重复:同一个 User 类型在 8 个仓库各定义一份,字段一改全端爆炸
API 契约漂移:API 仓库改了字段,Web 仓库不知道,上线就 500
UI 组件重复:按钮、对话框、消息气泡在 5 个端各实现一次,设计稿一变全端追改
发布协调难:types 仓库发了 v2,要等所有下游仓库升级,期间线上两个版本并存
CI 重复:8 个仓库各跑一遍 lint/typecheck/test,共享代码却无法增量

1.2 monorepo 的价值

┌─────────────────────────────────────────────────┐
│              IHUI-AI Monorepo                   │
│  ┌──────────────────────────────────────────┐   │
│  │  packages/ (共享:types/ui/auth/db/config)│   │
│  └──────────────────────────────────────────┘   │
│  ┌──────────────────────────────────────────┐   │
│  │  apps/ (8 端:web/api/ai-service/...)     │   │
│  └──────────────────────────────────────────┘   │
└─────────────────────────────────────────────────┘

单源真理:User 类型只在 packages/types 定义一次,8 端引用
原子提交:API 改字段 + Web 改调用,一个 commit 搞定
共享构建:lint/typecheck/test 配置在 packages/config,8 端复用
增量 CI:只重建受影响的包,Turborepo 自动判断

1.3 monorepo 的代价

不是没成本:

仓库体积大:clone 慢(但有 shallow clone 和 sparse checkout)
权限粒度粗:所有人能看所有代码(对开源项目反而是优势)
构建复杂:需要工具编排(Turborepo/Nx/Lerna)

权衡下来,8 端高度协同的 AI 平台,monorepo 收益远大于成本。

二、Turborepo vs Nx vs Lerna

2.1 选型对比

工具	语言	优势	劣势
Lerna	JS	老牌,文档多	已停止维护(2022 起)
Nx	TS	功能强,支持多语言,有缓存	学习曲线陡,配置重
Turborepo	Rust	极快,配置简单,远程缓存	生态新,插件少

IHUI-AI 选 Turborepo:

快:Rust 写的,任务编排 + 缓存比 Nx 快 3-5 倍
简单:turbo.json 一个文件搞定 pipeline,不需要学「generators/executors」概念
远程缓存:Vercel 免费托管,CI 命中缓存秒级完成
pnpm 原生支持:和我们的包管理器无缝集成

2.2 不选 Nx 的原因

Nx 适合「企业级 + 多语言 + 复杂依赖图」,但对我们这种纯 TS/Python 混合栈,Turborepo 已经够用。Nx 的 generator(代码生成)对我们价值不大,我们有自己的脚手架脚本。

三、pnpm workspace 配置

3.1 pnpm-workspace.yaml

# pnpm-workspace.yaml
packages:
  - "apps/*"
  - "packages/*"

这一行声明:所有 apps/ 和 packages/ 下的子目录都是 workspace 包,可以用 @ihui/<name> 互相引用。

3.2 catalog(统一版本)

pnpm 9.5+ 的 catalog 功能解决「同一依赖 8 个端版本不一致」的问题:

# pnpm-workspace.yaml
packages:
  - "apps/*"
  - "packages/*"

catalog:
  react: ^19.0.0
  react-dom: ^19.0.0
  typescript: ^5.6.0
  zod: ^3.23.0
  drizzle-orm: ^0.38.0
  tailwindcss: ^4.0.0

各包的 package.json 用 catalog: 引用:

{
  "name": "@ihui/web",
  "dependencies": {
    "react": "catalog:",
    "zod": "catalog:"
  }
}

改版本只改 pnpm-workspace.yaml 一处,所有包同步升级。

3.3 pnpm overrides(强制版本)

遇到安全漏洞要强制升级某个传递依赖:

// package.json (根)
{
  "pnpm": {
    "overrides": {
      "lodash": "^4.17.21"
    }
  }
}

四、8 端目录结构设计

IHUI-AI/
├── apps/
│   ├── web/              # Next.js 15 + React 19 + Tailwind 4
│   ├── api/              # Fastify 5 + Drizzle ORM 0.38
│   ├── ai-service/       # FastAPI + LangGraph + LiteLLM
│   ├── cli/              # Commander,发布到 npm
│   ├── desktop/          # Electron,复用 web 构建
│   ├── extension/        # Chrome MV3
│   ├── mobile-rn/        # React Native
│   └── miniapp-taro/     # Taro 4(微信小程序)
├── packages/
│   ├── types/            # 跨端共享 TypeScript 类型(API 契约)
│   ├── ui/               # 跨端共享 UI 组件(基于 shadcn/ui)
│   ├── database/         # Drizzle schema + RLS(共享给 api/ai-service)
│   ├── auth/             # 认证逻辑(共享给所有端)
│   ├── config/           # 共享配置(eslint/tailwind/tsconfig 预设)
│   ├── eslint-config/    # ESLint flat config 预设
│   └── tsconfig/         # TypeScript 配置预设
├── turbo.json            # Turborepo pipeline
├── pnpm-workspace.yaml   # pnpm workspace + catalog
└── package.json          # 根 package.json

设计原则:

apps/ 只写端特有代码:Web 的页面、CLI 的命令、Extension 的 background script
packages/ 只写共享代码:类型、UI、数据库 schema、认证
跨端契约在 packages/types:API 路由的请求/响应类型,所有端引用同一份

五、共享包:类型安全的命脉

5.1 packages/types 作为 single source of truth

// packages/types/src/api/chat.ts
import { z } from "zod";

// Zod schema = 运行时校验 + 类型推导
export const SendMessageRequest = z.object({
  conversationId: z.string(),
  content: z.string().min(1).max(8000),
  model: z.string().optional(),
});

// 类型自动从 schema 推导
export type SendMessageRequest = z.infer<typeof SendMessageRequest>;

export const SendMessageResponse = z.object({
  code: z.literal(0),
  message: z.string(),
  data: z.object({
    messageId: z.string(),
    content: z.string(),
    tokensUsed: z.number(),
  }),
});
export type SendMessageResponse = z.infer<typeof SendMessageResponse>;

5.2 API 端用 schema 校验

// apps/api/src/routes/chat.ts
import { SendMessageRequest } from "@ihui/types";

app.post("/api/chat/send", async (req, reply) => {
  const parsed = SendMessageRequest.safeParse(req.body);
  if (!parsed.success) {
    return reply.code(400).send({ code: 1, message: parsed.error.message });
  }
  // parsed.data 类型安全
  const result = await chatService.send(parsed.data);
  return { code: 0, message: "ok", data: result };
});

5.3 Web 端用同一份类型

// apps/web/src/api/chat.ts
import { SendMessageRequest, SendMessageResponse } from "@ihui/types";

export async function sendMessage(req: SendMessageRequest): Promise<SendMessageResponse> {
  const res = await fetch("/api/chat/send", {
    method: "POST",
    body: JSON.stringify(req),
  });
  return res.json();
}

API 改字段 → packages/types 编译报错 → Web 端立刻知道要改。编译期就发现契约漂移,而不是上线后 500。

六、构建编排:turbo.json

6.1 pipeline 定义

// turbo.json
{
  "$schema": "https://turbo.build/schema.json",
  "globalDependencies": ["tsconfig.json", ".env"],
  "pipeline": {
    "build": {
      "dependsOn": ["^build"],
      "outputs": ["dist/**", ".next/**", "build/**"],
      "env": ["NODE_ENV", "DATABASE_URL"]
    },
    "dev": {
      "cache": false,
      "persistent": true
    },
    "typecheck": {
      "dependsOn": ["^build"],
      "outputs": []
    },
    "lint": {
      "outputs": []
    },
    "test": {
      "dependsOn": ["^build"],
      "outputs": ["coverage/**"]
    },
    "db:generate": {
      "outputs": ["src/migrations/**"]
    },
    "db:migrate": {
      "cache": false
    }
  }
}

关键概念:

^build:先构建依赖项(packages/*),再构建自己(apps/*)
outputs:声明产物路径,Turborepo 据此做缓存
persistent: true:dev 是长任务,不缓存
cache: false:db:migrate 每次都真跑(改数据库不能靠缓存)

6.2 执行示例

# 全量构建(按依赖图并行)
pnpm turbo build
# 输出:
#  @ihui/types:build cache hit 1.2s
#  @ihui/database:build cache hit 2.1s
#  @ihui/web:build 45.3s
#  @ihui/api:build 12.4s
#  ...

# 只跑 typecheck(增量,只重建受影响的)
pnpm turbo typecheck --filter=@ihui/web

# 启动所有 dev server
pnpm turbo dev --parallel

6.3 远程缓存

# 登录 Vercel 免费 remote cache
npx turbo login
npx turbo link

# 之后 build 命中远程缓存,CI 时间从 8 分钟降到 90 秒

CI 上同一个 commit 第二次跑,几乎所有 task 都是 cache hit。

七、依赖管理

7.1 catalog 统一版本

见 3.2,所有共享依赖版本集中在 pnpm-workspace.yaml 的 catalog 字段。

7.2 端特有依赖就地声明

React Native 的 react-native、Taro 的 @tarojs/taro、Electron 的 electron,只在对应 apps/<端>/package.json 里声明,不污染共享包。

7.3 依赖图可视化

pnpm why react
# 列出谁依赖了 react,版本是否一致

# 检查重复依赖
pnpm list react --depth=10 --recursive

八、跨端类型安全

8.1 共享 schema(database + types)

packages/database 的 Drizzle schema 既是 SQL 生成源,也是类型源:

// packages/database/src/schema/users.ts
import { pgTable, varchar, timestamp, integer } from "drizzle-orm/pg-core";

export const users = pgTable("users", {
  id: varchar("id").primaryKey(),
  tenantId: varchar("tenant_id").notNull(),
  email: varchar("email").notNull(),
  roleId: integer("role_id").default(0).notNull(),
  createdAt: timestamp("created_at").defaultNow().notNull(),
});

// 类型自动推导
export type User = typeof users.$inferSelect;
export type NewUser = typeof users.$inferInsert;

API 端查询返回的 User 类型,Web 端拿到的也是同一个 User 类型——因为都从 @ihui/database 导入。

8.2 API 响应统一格式

所有 API 返回 { code, message, data }:

// packages/types/src/common.ts
export interface ApiResponse<T> {
  code: 0 | 1;
  message: string;
  data: T;
}

Web 端的 fetch wrapper 强制校验:

async function apiCall<T>(url: string, body?: unknown): Promise<T> {
  const res = await fetch(url, { /* ... */ });
  const json: ApiResponse<T> = await res.json();
  if (json.code !== 0) throw new Error(json.message);
  return json.data;
}

九、CI 优化

9.1 turbo prune(增量构建)

大型 monorepo 的 Docker 构建如果每次都 copy 全仓库,镜像层缓存会失效。turbo prune 只 copy 受影响的包:

# Dockerfile
FROM node:20 AS base
RUN corepack enable

FROM base AS installer
WORKDIR /app
COPY .gitignore .gitignore
COPY . .
# 只保留 web 及其依赖,剔除其他 7 端
RUN pnpm dlx turbo prune @ihui/web --docker

FROM base AS builder
WORKDIR /app
COPY --from=installer /app/out/json/ .
COPY --from=installer /app/out/pnpm-lock.yaml ./pnpm-lock.yaml
COPY --from=installer /app/out/full/ .
RUN pnpm install --frozen-lockfile
RUN pnpm turbo build --filter=@ihui/web...

FROM base AS runner
COPY --from=builder /app/apps/web/.next/standalone ./
EXPOSE 3000
CMD ["node", "server.js"]

Web 端 Docker 镜像从全量 2.3GB 降到 380MB,构建时间从 6 分钟降到 90 秒。

9.2 增量 task

CI 只跑受影响包的 task:

# 只 typecheck 改动过的包及其下游
pnpm turbo typecheck --filter=...[origin/main]

# 只测试改动过的包
pnpm turbo test --filter=...[origin/main]

9.3 并行 task

Turborepo 自动并行化无依赖关系的 task:

@ihui/types:build ──┬── @ihui/web:build
                    ├── @ihui/api:build
                    ├── @ihui/cli:build
                    └── @ihui/desktop:build  (并行)

8 个端的 build 在 8 核机器上并行跑,比串行快 4 倍。

十、常见陷阱

10.1 循环依赖

@ihui/types 依赖 @ihui/database,@ihui/database 又依赖 @ihui/types → 死循环。

解法:明确依赖方向。types 是最底层,不依赖任何业务包;database 依赖 types;apps 依赖两者。单向依赖图,不允许反向。

# 用 madge 检测循环依赖
npx madge --circular packages/ apps/ --extensions ts,tsx

10.2 版本漂移

没上 catalog 之前,8 个端的 zod 版本分别是 3.22 / 3.23 / 3.23 / 3.24...,某天 zod 3.24 改了 API,3.23 的端就炸了。

解法:catalog 强制统一(见 3.2)。所有包 zod: "catalog:",一处升级全端同步。

10.3 构建顺序错误

@ihui/web build 时依赖 @ihui/types 的产物,但 Turborepo 不知道顺序,可能先 build web 导致找不到类型。

解法:turbo.json 里 "dependsOn": ["^build"],Turborepo 会先 build 依赖项(^ 表示上游)。

10.4 workspace 协议误用

// 错误:用 ^ 范围,可能拉到 npm 上的旧版本
"dependencies": { "@ihui/types": "^1.0.0" }

// 正确:用 workspace 协议,强制用本地包
"dependencies": { "@ihui/types": "workspace:*" }

workspace:* 保证永远用本地 monorepo 的包,不会误拉 npm。

十一、IHUI-AI 实战配置

11.1 完整 turbo.json(精简版)

{
  "$schema": "https://turbo.build/schema.json",
  "globalDependencies": ["tsconfig.json", "pnpm-workspace.yaml"],
  "pipeline": {
    "build": {
      "dependsOn": ["^build"],
      "outputs": ["dist/**", ".next/**", "build/**", ".output/**"]
    },
    "dev": { "cache": false, "persistent": true },
    "typecheck": { "dependsOn": ["^build"], "outputs": [] },
    "lint": { "outputs": [] },
    "test": { "dependsOn": ["^build"], "outputs": ["coverage/**"] },
    "db:generate": { "outputs": ["src/migrations/**"] },
    "db:migrate": { "cache": false }
  }
}

11.2 package.json scripts(根)

{
  "scripts": {
    "dev": "turbo dev --parallel",
    "build": "turbo build",
    "typecheck": "turbo typecheck",
    "typecheck:full": "turbo typecheck --filter=...",
    "lint": "turbo lint",
    "test": "turbo test",
    "db:generate": "turbo db:generate",
    "db:migrate": "turbo db:migrate"
  }
}

pnpm dev 一条命令拉起 web(8801)+ api(8800)+ ai-service(8802)所有 dev server。

11.3 实测数据

操作	单仓库(估算)	monorepo(实测)	提升
全量 build	8 端串行 12min	并行 3min	4x
增量 typecheck	8 端全跑 4min	受影响 1 端 30s	8x
CI(命中缓存)	全量 8min	90s	5x
改 API 字段同步	手动通知 + 改 8 处	编译报错一处改	不可估量

十二、参考资料

Turborepo 官方文档
pnpm Workspace
pnpm Catalog
Nx 对比文档
turbo prune Docker
IHUI-AI 源码:turbo.json + pnpm-workspace.yaml + apps/* + packages/*

总结

8 端 monorepo 不是「为了 monorepo 而 monorepo」,是为了让 8 个端真正「同源」。Turborepo + pnpm Workspace + catalog 的组合,让我们能:

一处改,全端同步:packages/types 改字段,8 端编译报错,无遗漏
构建快:Turborepo 并行 + 远程缓存,CI 从 8 分钟降到 90 秒
版本统一:catalog 强制所有端用同一份依赖版本,杜绝漂移
增量交付:turbo prune 让单端 Docker 镜像只有 380MB

如果你也在做多端 AI 应用,monorepo 是值得前期投入的架构决策。短期看配置复杂,长期看每改一个字段省下的协调成本,远超初始投入。

至此,IHUI-AI 的工程实践系列博客已经覆盖了从 MCP 协议、PostgreSQL RLS、LangGraph 编排、LiteLLM Gateway 到 Monorepo 构建的完整技术栈。回到第一篇 8 端同源架构可以看到,这些技术选型共同支撑了一个「8 端同源、340 表、176 模型」的全栈 AI 平台。

This article was originally published on the IHUI AI Blog. Follow us on GitHub for more AI engineering content.

LiteLLM 适配 176 个大模型:构建统一 LLM Gateway 实践

李春川 — Tue, 28 Jul 2026 05:38:37 +0000

LiteLLM 适配 176 个大模型:构建统一 LLM Gateway 实践

做 AI 应用最痛苦的事之一:每家 LLM 厂商的 SDK 都不一样。OpenAI 一套、Anthropic 一套、Gemini 一套、国内厂商又各一套。你的代码里到处是 if model.startswith("gpt") 的分支。LiteLLM 用一个 OpenAI 兼容的接口,把 176 个模型统一成一种调用方式。本文是 IHUI-AI 在 apps/ai-service/app/core/llm_gateway.py 落地 LLM Gateway 的工程总结。

一、为什么需要 LLM Gateway

1.1 没有 Gateway 的痛苦

# 早期代码:到处是 if-else
def call_llm(model: str, messages: list):
    if model.startswith("gpt"):
        from openai import OpenAI
        client = OpenAI()
        return client.chat.completions.create(model=model, messages=messages)
    elif model.startswith("claude"):
        import anthropic
        client = anthropic.Anthropic()
        return client.messages.create(model=model, messages=messages)
    elif model.startswith("gemini"):
        import google.generativeai as genai
        genai.configure(api_key=os.getenv("GEMINI_KEY"))
        # 完全不同的 API 形状
        ...
    elif model.startswith("deepseek"):
        # 又一个 SDK
        ...
    # 176 个模型 = 176 套分支

问题:

接口不统一:每家 SDK 的入参/出参/流式格式都不同
错误处理散乱:OpenAI 抛 RateLimitError,Anthropic 抛 APITimeoutError,没法统一 catch
成本统计难:每家计价单位不同(token 定义都不一样)
故障转移难:GPT-4 限流了想切 Claude,要写一堆切换逻辑

1.2 LLM Gateway 的四大价值

价值	说明
统一接口	所有模型用 OpenAI 兼容格式调用,一套代码全覆盖
成本控制	统一计费、预算告警、租户配额
故障转移	主模型挂了自动切备用,业务无感
审计合规	所有调用记录:谁、何时、调了什么、花了多少、结果

二、LiteLLM 设计原理

LiteLLM 的核心思想:所有 LLM 都能被映射成 OpenAI Chat Completions API。

2.1 三层架构

┌─────────────────────────────────────┐
│  你的应用(OpenAI 兼容调用)        │
└──────────────┬──────────────────────┘
               │ uniform interface
┌──────────────▼──────────────────────┐
│  LiteLLM Router / Proxy             │
│  - 路由策略                          │
│  - 成本控制                          │
│  - 故障转移                          │
└──────────────┬──────────────────────┘
               │ translate
┌──────────────▼──────────────────────┐
│  各厂商 SDK / API                   │
│  OpenAI | Anthropic | Gemini | ...  │
└─────────────────────────────────────┘

2.2 统一调用示例

from litellm import completion

# 调 OpenAI
response = completion(model="gpt-4o", messages=[{"role": "user", "content": "Hi"}])

# 调 Anthropic(完全一样的代码,只改 model 名)
response = completion(model="claude-3-5-sonnet-20241022", messages=[...])

# 调 Gemini
response = completion(model="gemini/gemini-1.5-pro", messages=[...])

# 调 DeepSeek
response = completion(model="deepseek/deepseek-chat", messages=[...])

# 调本地 Ollama
response = completion(model="ollama/llama3", messages=[...], api_base="http://localhost:11434")

返回的 response 对象结构完全一致(choices[0].message.content),无需关心底层差异。

三、176 个模型的适配分类

LiteLLM 支持的 176 个模型按厂商分类:

3.1 国际厂商

厂商	model 前缀	代表模型
OpenAI	`gpt-*`	gpt-4o, gpt-4o-mini, o1-preview
Anthropic	`claude-*`	claude-3-5-sonnet, claude-3-opus
Google	`gemini/*`	gemini-1.5-pro, gemini-1.5-flash
Mistral	`mistral/*`	mistral-large, mistral-small
Cohere	`command-r-*`	command-r, command-r-plus
Azure	`azure/<deploy>`	Azure 托管的 OpenAI
AWS	`bedrock/*`	Bedrock 托管的多厂商

3.2 国内厂商(IHUI-AI 重点)

厂商	model 前缀	代表模型
DeepSeek	`deepseek/*`	deepseek-chat, deepseek-coder
通义 Qwen	`dashscope/*`	qwen-max, qwen-plus, qwen-turbo
智谱 GLM	`zhipu/*`	glm-4, glm-4-plus, glm-4-flash
字节豆包	`volcengine/*`	doubao-pro, doubao-lite
月之暗面	`moonshot/*`	moonshot-v1-8k, moonshot-v1-32k
MiniMax	`minimax/*`	abab6.5, abab6.5s
讯飞星火	`spark/*`	spark-v3.5, spark-v4
百度文心	`wenxin/*`	ernie-bot-4, ernie-bot-turbo

3.3 开源自托管

厂商	model 前缀	代表模型
Ollama	`ollama/*`	llama3, qwen2, mistral
vLLM	`vllm/*`	任意 HuggingFace 模型
LlamaCpp	`llamacpp/*`	本地 GGUF 模型
LM Studio	`lm_studio/*`	LM Studio 桌面端

国内场景下,IHUI-AI 默认推荐 DeepSeek(性价比高)+ Qwen(阿里云生态)+ GLM(国产合规),配合 Ollama 做本地开发。

四、路由策略

LLM Gateway 的核心是 router,决定「这次调用走哪个模型」。

4.1 Router 配置

# apps/ai-service/app/core/llm_gateway.py
from litellm import Router

router = Router(model_list=[
    {
        "model_name": "default",  # 业务侧用的别名
        "litellm_params": {
            "model": "deepseek/deepseek-chat",
            "api_key": os.getenv("DEEPSEEK_KEY"),
        },
    },
    {
        "model_name": "default",  # 同名 = 同组,故障转移候选
        "litellm_params": {
            "model": "gpt-4o-mini",
            "api_key": os.getenv("OPENAI_KEY"),
        },
    },
    {
        "model_name": "premium",
        "litellm_params": {
            "model": "claude-3-5-sonnet-20241022",
            "api_key": os.getenv("ANTHROPIC_KEY"),
        },
    },
    {
        "model_name": "premium",
        "litellm_params": {
            "model": "gpt-4o",
            "api_key": os.getenv("OPENAI_KEY"),
        },
    },
])

# 业务侧:用别名,router 决定实际走哪个
response = router.completion(model="default", messages=[...])

4.2 路由策略对比

策略	配置	适合场景
成本优先	`routing_strategy="cost-based"`	免费用户、内部测试
性能优先	`routing_strategy="latency-based-routing"`	付费用户、实时场景
故障转移	`fallbacks=["premium"]`	高可用要求
地域优先	自定义权重	合规、延迟优化
轮询	`routing_strategy="simple-shuffle"`	负载均衡

4.3 故障转移实战

router = Router(
    model_list=[...],
    fallbacks=[
        {"default": ["premium"]},   # default 全挂 → 切 premium
        {"premium": ["default"]},   # premium 全挂 → 降级 default
    ],
    routing_strategy="latency-based-routing",
    num_retries=2,
    retry_after=1,
    timeout=30,
)

# DeepSeek 限流 → 自动切 GPT-4o-mini
response = await router.acompletion(model="default", messages=[...])

实测:DeepSeek 高峰期限流率约 5%,有 fallback 后业务侧完全无感。

五、成本优化

5.1 缓存

相同 prompt + model 的结果缓存(Redis):

from litellm import completion
from litellm.caching import Cache

litellm.cache = Cache(
    type="redis",
    host=os.getenv("REDIS_HOST"),
    port=6379,
    ttl=3600,  # 1 小时
)

# 相同输入第二次走缓存,0 成本
response = completion(model="gpt-4o", messages=[...], caching=True)

适合:知识库 QA(同一问题反复问)、分类标签(同一段文本分类稳定)。不适合:创意写作、对话(每次都不同)。

5.2 批处理

多个独立请求合并成一个 batch,部分厂商支持降价 50%:

from litellm import batch_completion

responses = batch_completion(
    model="gpt-4o-mini",
    messages=[
        [{"role": "user", "content": "翻译: Hello"}],
        [{"role": "user", "content": "翻译: World"}],
        [{"role": "user", "content": "翻译: Goodbye"}],
    ],
)

OpenAI 的 Batch API 异步批处理,24 小时内返回,价格减半。

5.3 模型降级

简单任务用小模型,复杂任务才用大模型:

def smart_route(messages: list) -> str:
    total_tokens = sum(estimate_tokens(m["content"]) for m in messages)
    last_msg = messages[-1]["content"]

    # 简单问候/闲聊 → 小模型
    if total_tokens < 100 and len(last_msg) < 50:
        return "default"  # deepseek-chat / gpt-4o-mini

    # 长文本/复杂推理 → 大模型
    if total_tokens > 5000 or "分析" in last_msg or "代码" in last_msg:
        return "premium"  # claude / gpt-4o

    return "default"

model = smart_route(messages)
response = await router.acompletion(model=model, messages=messages)

实测:接入 smart_route 后,平均成本下降 60%,因为 80% 的请求其实只需要小模型。

5.4 token 估算

import tiktoken

def estimate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    pricing = {
        "gpt-4o":           {"input": 2.5e-6, "output": 10e-6},
        "gpt-4o-mini":      {"input": 0.15e-6, "output": 0.6e-6},
        "claude-3-5-sonnet":{"input": 3e-6, "output": 15e-6},
        "deepseek-chat":    {"input": 0.14e-6, "output": 0.28e-6},
    }
    p = pricing.get(model, {"input": 1e-6, "output": 2e-6})
    return input_tokens * p["input"] + output_tokens * p["output"]

DeepSeek 比 GPT-4o 便宜约 18 倍,这是国内 SaaS 的成本优势所在。

六、流式响应(SSE)+ 客户端断连

6.1 流式输出

from litellm import acompletion

async def stream_chat(model: str, messages: list):
    response = await acompletion(
        model=model,
        messages=messages,
        stream=True,
    )
    async for chunk in response:
        delta = chunk.choices[0].delta.content
        if delta:
            yield f"data: {json.dumps({'content': delta})}\n\n"
    yield "data: [DONE]\n\n"

6.2 客户端断连处理

用户关浏览器时,SSE 连接断开,但 LLM 还在生成(继续计费)。要监听断开事件取消请求:

from fastapi import Request
from fastapi.responses import StreamingResponse
import asyncio

@app.post("/chat/stream")
async def chat_stream(request: Request, body: ChatBody):
    async def event_stream():
        task = asyncio.create_task(stream_chat(body.model, body.messages))
        try:
            async for chunk in task:
                if await request.is_disconnected():
                    task.cancel()  # 客户端断了,取消 LLM 调用
                    break
                yield chunk
        except asyncio.CancelledError:
            task.cancel()
            raise

    return StreamingResponse(event_stream(), media_type="text/event-stream")

这一步能省下 10-20% 的 token 成本(用户中途关掉的请求不会继续跑)。

七、限流与配额

7.1 per-tenant / per-user / per-model

from litellm import Router

router = Router(
    model_list=[...],
    router_budget_limit={
        "default": {"tenant_free": 1000, "tenant_pro": 100000},  # tokens/day
    },
    routing_strategy="usage-based-routing-v2",
)

# 每次调用前 router 检查租户今日用量
response = await router.acompletion(
    model="default",
    messages=[...],
    metadata={"tenant_id": tenant_id, "tenant_tier": "free"},
)

超额时 router 抛 BudgetExceededError,业务侧返回 429。

7.2 速率限制

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@app.post("/chat")
@limiter.limit("30/minute")  # 每 IP 每分钟 30 次
async def chat(request: Request, body: ChatBody):
    ...

八、IHUI-AI 实战:llm_gateway.py

apps/ai-service/app/core/
├── llm_gateway.py        # LiteLLM router 封装
├── llm_config.py         # 模型配置(从 DB 加载)
├── llm_cost.py           # 成本统计
└── llm_audit.py          # 审计日志

8.1 统一入口

# apps/ai-service/app/core/llm_gateway.py
from litellm import Router
from litellm.exceptions import BudgetExceededError, RateLimitError
import structlog

logger = structlog.get_logger()

class LLMGateway:
    def __init__(self):
        self.router = Router(model_list=self._load_models())
        self.router.fallbacks = self._load_fallbacks()

    async def chat(
        self,
        messages: list,
        model_alias: str = "default",
        tenant_id: str = "",
        user_id: str = "",
        stream: bool = False,
    ) -> dict:
        metadata = {"tenant_id": tenant_id, "user_id": user_id}
        try:
            response = await self.router.acompletion(
                model=model_alias,
                messages=messages,
                stream=stream,
                metadata=metadata,
            )
            # 记录审计
            await self._audit(tenant_id, user_id, model_alias, response)
            return response
        except BudgetExceededError:
            logger.warning("budget_exceeded", tenant=tenant_id)
            raise HTTPException(429, "今日额度已用完")
        except RateLimitError:
            logger.warning("rate_limited", model=model_alias)
            raise HTTPException(429, "模型繁忙,请稍后")

    async def _audit(self, tenant_id, user_id, model, response):
        # 写入 llm_calls 表,记录 token / 成本 / 状态
        await db.execute(
            "INSERT INTO llm_calls (tenant_id, user_id, model, input_tokens, output_tokens, cost_usd, status) VALUES ($1,$2,$3,$4,$5,$6,$7)",
            tenant_id, user_id, model,
            response.usage.prompt_tokens,
            response.usage.completion_tokens,
            estimate_cost(model, response.usage.prompt_tokens, response.usage.completion_tokens),
            "success",
        )

gateway = LLMGateway()

8.2 模型配置从 DB 加载

admin 后台可以动态增删模型,无需重启:

async def _load_models(self) -> list:
    rows = await db.fetch("SELECT * FROM llm_models WHERE enabled = TRUE")
    return [{
        "model_name": row["alias"],  # 业务别名,如 "default"
        "litellm_params": {
            "model": row["provider_model"],  # 实际模型名,如 "deepseek/deepseek-chat"
            "api_key": decrypt(row["api_key_encrypted"]),
            "api_base": row.get("api_base"),
        },
    } for row in rows]

九、监控:token / 错误率 / 延迟 / 成本

9.1 监控仪表盘

-- 今日各模型 token 消耗
SELECT model,
       COUNT(*) AS calls,
       SUM(input_tokens) AS input_toks,
       SUM(output_tokens) AS output_toks,
       SUM(cost_usd) AS cost
FROM llm_calls
WHERE created_at::date = CURRENT_DATE
GROUP BY model ORDER BY cost DESC;

-- 错误率
SELECT model,
       COUNT(*) FILTER (WHERE status != 'success') * 100.0 / COUNT(*) AS error_rate
FROM llm_calls
WHERE created_at > NOW() - INTERVAL '1 hour'
GROUP BY model;

-- P95 延迟
SELECT model,
       PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY duration_ms) AS p95
FROM llm_calls
GROUP BY model;

9.2 告警

单租户 1 小时成本 > $10 → 告警(可能有滥用)
某模型错误率 > 10% → 告警(可能厂商故障,触发 fallback)
P95 延迟 > 10s → 告警(模型过载)

十、与 MCP 协议的协同

LiteLLM 把「选哪个模型」收敛成 model="default",而 MCP 把「调哪个工具」收敛成 tools=[...]。两者结合:

async def smart_agent(messages: list, tenant_id: str):
    # 1. 用 LiteLLM 选模型(成本/性能路由)
    model = await gateway.pick_model(messages, tenant_id)

    # 2. 用 MCP 拉工具(按租户权限)
    tools = await mcp_registry.discover_tools(tenant_id)

    # 3. 统一调用
    response = await gateway.chat(
        messages=messages,
        model_alias=model,
        tools=tools,  # LiteLLM 自动转成厂商的 tool calling 格式
    )
    return response

LiteLLM 内部会把 OpenAI 格式的 tools 翻译成 Anthropic 的 tools、Gemini 的 function_declarations 等,业务侧只写一份。

更进一步:可以把「模型选择」本身做成一个 MCP 工具,让 agent 自己根据任务复杂度选模型(简单任务选便宜模型,复杂任务选强模型)。

十一、参考资料

LiteLLM 官方文档
LiteLLM 支持的 176 个模型清单
LiteLLM Router 策略
LiteLLM Proxy Server
IHUI-AI 源码:apps/ai-service/app/core/llm_gateway.py

总结

LLM Gateway 不是「多此一举」,而是 AI 应用从 demo 走向生产的必经之路。LiteLLM 用 OpenAI 兼容接口统一 176 个模型,解决了:

接口碎片化:一套代码调所有厂商
成本失控:缓存 + 批处理 + 模型降级,成本砍 60%
故障频发:fallback 自动切备用,业务无感
审计缺失:每次调用记录 token/成本,精准计费

IHUI-AI 的实践配置:DeepSeek(默认,性价比)+ GPT-4o-mini(fallback)+ Claude 3.5(premium),配合 Redis 缓存和 smart_route,单次对话平均成本控制在 $0.001 以内。

下一篇讲 Monorepo 8 端同步开发,看所有这些服务如何在一个仓库里协同构建。

This article was originally published on the IHUI AI Blog. Follow us on GitHub for more AI engineering content.

LangGraph Agent 编排模式:6 种生产级 Multi-Agent 架构

李春川 — Tue, 28 Jul 2026 05:38:33 +0000

LangGraph Agent 编排模式:6 种生产级 Multi-Agent 架构

单个 LLM agent 能力有限:它会被长任务搞晕、会在多步推理中失忆、会把所有职责搅成一团。生产级的 AI 系统需要多个 agent 分工协作——但「多个 agent 怎么协作」本身就是个难题。LangGraph 把 agent 编排建模成「状态图」,让复杂协作变得可调试、可持久化、可观测。本文是 IHUI-AI 在 apps/ai-service/app/graphs/ 落地 6 种编排模式的工程总结。

一、LangGraph 与 LangChain 的关系

很多人分不清 LangChain 和 LangGraph。

1.1 LangChain 是「链」

LangChain 的核心抽象是 Chain——线性的步骤序列:

[输入] → [Prompt 模板] → [LLM] → [输出解析] → [输出]

适合简单任务,但遇到「根据结果决定下一步走哪」「循环直到满足条件」就力不从心。

1.2 LangGraph 是「图」

LangGraph 把执行流建模成有向图:

节点(Node):一个函数,接收 state,返回 state 更新
边(Edge):节点间的转移,可以是固定的或带条件的
状态(State):在节点间流转的共享数据,用 TypedDict 定义

       ┌─────────┐
       │  start  │
       └────┬────┘
            ▼
       ┌─────────┐    条件    ┌─────────┐
       │ router  │ ─────────▶ │ agent_a │
       └────┬────┘            └─────────┘
            │ 条件
            ▼
       ┌─────────┐
       │ agent_b │
       └─────────┘

图天然支持:分支、循环、并行、汇聚——这正是 multi-agent 协作需要的。

1.3 为什么选 LangGraph

需求	LangChain Chain	LangGraph
线性流程	✅	✅
条件分支	❌	✅
循环(ReAct)	需 hack	✅ 原生
并行 fan-out	❌	✅
状态持久化	❌	✅
人机协同(HITL)	❌	✅
流式事件	部分	✅ 完善

二、State 设计:编排的基础

所有 LangGraph 编排都从 State 定义开始。

2.1 TypedDict + Annotated

# apps/ai-service/app/graphs/state.py
from typing import TypedDict, Annotated, Literal
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    # add_messages reducer:列表追加而非覆盖
    messages: Annotated[list[BaseMessage], add_messages]
    # 普通字段:直接覆盖
    tenant_id: str
    user_id: str
    # 业务字段
    task: str
    research_notes: str
    draft: str
    # 路由控制
    next: str
    iteration: int
    status: Literal["running", "done", "error"]

关键点:

Annotated[list, add_messages] 用 reducer,新消息追加而不是覆盖
普通字段默认是「覆盖」语义
next 字段用于条件边路由

2.2 Channel(高级)

复杂场景下可以用 Channel 做更细粒度的状态管理,比如「每个 agent 自己的私有状态」。本文聚焦基础 TypedDict,Channel 留给进阶读者。

三、6 种编排模式

3.1 模式一:序列(Sequential)

最简单的模式:agent 依次执行,前者输出是后者输入。

from langgraph.graph import StateGraph, END

def researcher(state: AgentState) -> dict:
    """调研 agent:收集资料。"""
    notes = call_llm(f"研究这个主题: {state['task']}")
    return {"research_notes": notes}

def writer(state: AgentState) -> dict:
    """写作 agent:基于资料写初稿。"""
    draft = call_llm(f"基于以下资料写文章:\n{state['research_notes']}")
    return {"draft": draft}

def reviewer(state: AgentState) -> dict:
    """审阅 agent:检查并定稿。"""
    final = call_llm(f"审阅并润色:\n{state['draft']}")
    return {"draft": final, "status": "done"}

graph = StateGraph(AgentState)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
graph.add_node("reviewer", reviewer)

graph.set_entry_point("researcher")
graph.add_edge("researcher", "writer")
graph.add_edge("writer", "reviewer")
graph.add_edge("reviewer", END)

app = graph.compile()

适合:流水线任务(调研→写作→审阅)。不适合:需要分支判断的场景。

3.2 模式二:并行(Parallel fan-out / fan-in)

多个 agent 同时工作,结果汇聚。

from langgraph.graph import StateGraph, END
import asyncio

def fan_out_research(state: AgentState) -> dict:
    """并行调研多个子主题。"""
    async def research_topic(topic: str) -> str:
        return await async_call_llm(f"研究: {topic}")

    topics = extract_subtopics(state["task"])
    results = asyncio.run(asyncio.gather(*[research_topic(t) for t in topics]))
    return {"research_notes": "\n---\n".join(results)}

def synthesize(state: AgentState) -> dict:
    """综合所有调研结果。"""
    draft = call_llm(f"综合以下调研写综述:\n{state['research_notes']}")
    return {"draft": draft, "status": "done"}

graph = StateGraph(AgentState)
graph.add_node("fan_out_research", fan_out_research)
graph.add_node("synthesize", synthesize)
graph.set_entry_point("fan_out_research")
graph.add_edge("fan_out_research", "synthesize")
graph.add_edge("synthesize", END)
app = graph.compile()

LangGraph 也支持用 add_node 注册多个节点 + add_edge 从同一源点指向多个目标,框架自动并行执行。

适合:信息聚合(多源检索、多视角分析)。

3.3 模式三:路由(Conditional routing)

根据输入特征走不同分支。

def router(state: AgentState) -> str:
    """根据任务类型路由到不同 agent。"""
    task = state["task"].lower()
    if "代码" in task or "code" in task:
        return "code_agent"
    elif "翻译" in task or "translate" in task:
        return "translate_agent"
    else:
        return "chat_agent"

def code_agent(state: AgentState) -> dict:
    return {"messages": [ai_msg("代码 agent 处理中...")]}

def translate_agent(state: AgentState) -> dict:
    return {"messages": [ai_msg("翻译 agent 处理中...")]}

def chat_agent(state: AgentState) -> dict:
    return {"messages": [ai_msg("通用 chat agent...")]}

graph = StateGraph(AgentState)
graph.add_node("code_agent", code_agent)
graph.add_node("translate_agent", translate_agent)
graph.add_node("chat_agent", chat_agent)

graph.set_conditional_entry_point(
    router,
    {"code_agent": "code_agent", "translate_agent": "translate_agent", "chat_agent": "chat_agent"},
)
graph.add_edge("code_agent", END)
graph.add_edge("translate_agent", END)
graph.add_edge("chat_agent", END)
app = graph.compile()

适合:任务分发(客服路由、工单分类)。

3.4 模式四:协作(Collaborative)

多个 agent 共享 state,互相补充。

def coder(state: AgentState) -> dict:
    """写代码。"""
    code = call_llm(f"写代码: {state['task']}")
    return {"draft": code, "next": "critic"}

def critic(state: AgentState) -> dict:
    """挑刺。"""
    feedback = call_llm(f"审查这段代码的问题:\n{state['draft']}")
    if "没问题" in feedback:
        return {"status": "done", "next": END}
    return {"research_notes": feedback, "next": "coder"}  # 回去重写

def should_continue(state: AgentState) -> str:
    return state.get("next", END)

graph = StateGraph(AgentState)
graph.add_node("coder", coder)
graph.add_node("critic", critic)
graph.set_entry_point("coder")
graph.add_conditional_edges("coder", should_continue, {"coder": "coder", "critic": "critic"})
graph.add_conditional_edges("critic", should_continue, {"coder": "coder", END: END})
app = graph.compile()

特点:带循环,coder 和 critic 反复迭代直到 critic 满意。LangGraph 原生支持循环,这是 LangChain Chain 做不到的。

3.5 模式五:监督(Supervisor)

一个 supervisor agent 决定下一步调用哪个 worker。

def supervisor(state: AgentState) -> dict:
    """监督者:决定下一步交给谁。"""
    prompt = f"""你是任务协调者。当前任务: {state['task']}
    已完成步骤: {state.get('research_notes', '无')}
    可选 worker: [researcher, coder, tester, FINISH]
    返回 worker 名或 FINISH。"""
    decision = call_llm(prompt).strip()
    return {"next": decision}

def researcher(state: AgentState) -> dict:
    return {"research_notes": call_llm(f"研究: {state['task']}")}

def coder(state: AgentState) -> dict:
    return {"draft": call_llm(f"写代码: {state['task']}")}

def tester(state: AgentState) -> dict:
    return {"research_notes": call_llm(f"测试: {state['draft']}")}

def route(state: AgentState) -> str:
    nxt = state["next"]
    return nxt if nxt != "FINISH" else END

graph = StateGraph(AgentState)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher)
graph.add_node("coder", coder)
graph.add_node("tester", tester)

graph.set_entry_point("supervisor")
graph.add_conditional_edges("supervisor", route, {
    "researcher": "researcher", "coder": "coder", "tester": "tester", END: END
})
# 每个 worker 完成后回到 supervisor
for w in ["researcher", "coder", "tester"]:
    graph.add_edge(w, "supervisor")
app = graph.compile()

这是 LangGraph 官方推荐的 multi-agent 模式。supervisor 像「项目经理」,worker 像「专家」,适合复杂多步任务。

3.6 模式六:层级(Hierarchical)

supervisor 之上还有 supervisor,形成树状结构。

              [Top Supervisor]
              /       |       \
        [Sub-Sup1] [Sub-Sup2] [Sub-Sup3]
         /    \      |          |    \
      [R]   [C]    [T]        [W]  [D]

# 顶层 supervisor 决定交给哪个子团队
def top_supervisor(state: AgentState) -> dict:
    phase = state.get("iteration", 0)
    if phase == 0:
        return {"next": "research_team", "iteration": 1}
    elif phase == 1:
        return {"next": "dev_team", "iteration": 2}
    else:
        return {"next": "FINISH"}

# 子团队的 supervisor 各自管理自己的 worker
def research_team_sup(state: AgentState) -> dict:
    # 内部可以用子图(subgraph)
    return {"research_notes": "团队调研结果", "next": "top"}

graph = StateGraph(AgentState)
graph.add_node("top_supervisor", top_supervisor)
graph.add_node("research_team", research_team_sup)
graph.add_node("dev_team", lambda s: {"draft": "代码", "next": "top"})
graph.set_entry_point("top_supervisor")
graph.add_conditional_edges("top_supervisor", lambda s: s["next"], {
    "research_team": "research_team", "dev_team": "dev_team", "FINISH": END
})
graph.add_edge("research_team", "top_supervisor")
graph.add_edge("dev_team", "top_supervisor")
app = graph.compile()

适合:大型项目(研究团队→开发团队→测试团队,各司其职)。

四、持久化:Checkpointer + Memory

生产环境的 agent 必须能「断点续跑」。

4.1 Checkpointer

LangGraph 的 Checkpointer 在每个节点执行后保存 state 快照:

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.checkpoint.memory import MemorySaver

# 生产用 Postgres,开发用 Memory
checkpointer = PostgresSaver.from_conn_string(DATABASE_URL)
# checkpointer = MemorySaver()

app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["critic"],  # 在 critic 前暂停(人机协同)
)

# 用 thread_id 关联会话
config = {"configurable": {"thread_id": "session-abc123"}}
result = await app.ainvoke(initial_state, config=config)

# 之后可以恢复
resumed = await app.ainvoke(None, config=config)  # 从断点继续

4.2 实战:长任务断点续跑

IHUI-AI 的「深度研究」任务可能跑 10 分钟,用户中途关浏览器,后台继续跑,重开页面能看到进度:

# 前端轮询接口
@app.get("/api/tasks/{task_id}/state")
async def get_task_state(task_id: str):
    state = await app.aget_state({"configurable": {"thread_id": task_id}})
    return {
        "current_node": state.next,
        "values": state.values,
        "done": state.next == END,
    }

五、人机协同(Human-in-the-loop)

某些决策必须人类介入(如「这个操作要花 100 美元,确认?」)。

5.1 interrupt_before

app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["execute_expensive_op"],
)

# 跑到 execute_expensive_op 前会暂停
await app.ainvoke(state, config={"configurable": {"thread_id": tid}})

# 等人类确认
@app.post("/api/tasks/{tid}/approve")
async def approve(tid: str):
    # 人类点了「确认」,继续执行
    await app.ainvoke(None, config={"configurable": {"thread_id": tid}})

5.2 动态 interrupt

from langgraph.types import interrupt

def risky_action(state: AgentState) -> dict:
    cost = estimate_cost(state["task"])
    if cost > 10:
        # 运行时动态暂停,问人类
        approved = interrupt({"question": f"此操作花费 ${cost},确认?", "cost": cost})
        if not approved:
            return {"status": "error", "next": END}
    # 执行
    return {"status": "done"}

六、流式输出(Streaming events)

LLM 响应要流式给前端,否则用户等 30 秒看不到任何东西会以为崩了。

async def stream_response(task_id: str):
    config = {"configurable": {"thread_id": task_id}}
    async for event in app.astream_events(initial_state, config=config, version="v2"):
        kind = event["event"]
        if kind == "on_chat_model_stream":
            # LLM token 流
            chunk = event["data"]["chunk"].content
            yield f"data: {json.dumps({'type': 'token', 'content': chunk})}\n\n"
        elif kind == "on_chain_start":
            yield f"data: {json.dumps({'type': 'node_start', 'node': event['name']})}\n\n"
        elif kind == "on_chain_end":
            yield f"data: {json.dumps({'type': 'node_end', 'node': event['name']})}\n\n"

前端用 EventSource(SSE)接收,既显示 token 流又显示当前在哪个节点。

七、错误处理与重试

agent 调用 LLM 会失败(限流、超时、JSON 解析错)。每个节点要包重试:

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
def robust_node(state: AgentState) -> dict:
    try:
        result = call_llm(state["task"])
        return {"draft": result}
    except json.JSONDecodeError:
        # LLM 返回的 JSON 畸形,加修复提示重试
        return {"draft": "", "error": "JSON parse failed"}
    except Exception as e:
        logger.error("node_failed", error=str(e), state=state)
        raise

# 全局 fallback:图级别的错误兜底
app = graph.compile(checkpointer=checkpointer)
try:
    result = await app.ainvoke(state, config=config)
except Exception as e:
    # 标记任务失败,记录到 DB
    await mark_task_failed(task_id, str(e))

八、IHUI-AI 实战:apps/ai-service/app/graphs/

apps/ai-service/app/graphs/
├── __init__.py
├── state.py              # AgentState 定义
├── chat_graph.py         # 聊天(supervisor 模式)
├── research_graph.py     # 深度研究(层级模式)
├── task_graph.py         # 任务执行(协作模式)
└── tools/                # 各 agent 用的工具
    ├── search.py
    ├── code.py
    └── write.py

8.1 聊天 graph(supervisor)

聊天场景下,supervisor 决定:这次对话是直接回 / 查知识库 / 调工具 / 转 agent:

async def build_chat_graph(tenant_id: str, user_role: str):
    tools = await load_tools_for_role(user_role)  # MCP 工具
    model = init_model("gpt-4o")

    def supervisor(state: AgentState) -> dict:
        last_msg = state["messages"][-1].content
        if need_search(last_msg):
            return {"next": "search_agent"}
        elif need_tool(last_msg):
            return {"next": "tool_agent"}
        else:
            return {"next": "direct_reply"}

    # ... 构建图,绑定 MCP 工具到 tool_agent
    return graph.compile(checkpointer=postgres_checkpointer)

8.2 深度研究 graph(层级)

「帮我研究 XX 主题,写一份报告」这种长任务用层级模式:

Top supervisor:规划阶段(调研→写作→审阅)
Research team:fan-out 多个 sub-topic 并行检索
Writer team:综合调研写报告
Reviewer:审阅打回或定稿

实测一个 10 分钟的研究任务,token 消耗约 50k,成本约 $0.3。

九、性能调优

9.1 并发控制

并行 fan-out 时,限制并发数避免打爆 LLM API:

import asyncio

async def fan_out_limited(tasks: list, max_concurrent: int = 5):
    sem = asyncio.Semaphore(max_concurrent)
    async def bounded(t):
        async with sem:
            return await t
    return await asyncio.gather(*[bounded(t) for t in tasks])

9.2 token 估算

import tiktoken
enc = tiktoken.encoding_for_model("gpt-4o")

def estimate_tokens(text: str) -> int:
    return len(enc.encode(text))

# 在 supervisor 里预估下一步成本
def supervisor(state: AgentState) -> dict:
    input_tokens = sum(estimate_tokens(m.content) for m in state["messages"])
    if input_tokens > 100_000:
        return {"next": "summarize"}  # 压缩历史
    return {"next": "continue"}

9.3 缓存

相同子任务的结果缓存到 Redis:

import hashlib, json

async def cached_call(prompt: str, ttl=3600):
    key = f"llm:{hashlib.sha256(prompt.encode()).hexdigest()}"
    if cached := await redis.get(key):
        return json.loads(cached)
    result = await call_llm(prompt)
    await redis.setex(key, ttl, json.dumps(result))
    return result

十、参考资料

LangGraph 官方文档
LangGraph Multi-Agent Tutorial
LangChain Python
IHUI-AI 源码:apps/ai-service/app/graphs/

总结

LangGraph 把 multi-agent 编排从「if-else 意大利面」变成「可可视化、可持久化、可断点续跑的状态图」。6 种模式覆盖了绝大多数场景:

序列:流水线(调研→写作→审阅)
并行:多源聚合(多主题检索)
路由:任务分发(按类型走不同 agent)
协作:带循环的迭代(coder↔critic)
监督:项目经理模式(supervisor 调度 worker)
层级:大型项目(团队树)

IHUI-AI 的实践表明,supervisor 模式是性价比最高的起点:一个 LLM 当调度器,几个专精 agent 当 worker,加上 Postgres checkpointer 做断点续跑,就能撑住生产负载。等业务复杂了再演进到层级模式。

下一篇讲 LiteLLM 适配 176 个大模型,看 agent 背后的 LLM 调用如何统一。

This article was originally published on the IHUI AI Blog. Follow us on GitHub for more AI engineering content.

PostgreSQL 多租户行级安全:340 张表的生产级 RLS 实践

李春川 — Tue, 28 Jul 2026 05:38:28 +0000

PostgreSQL 多租户行级安全:340 张表的生产级 RLS 实践

SaaS 应用的生死线是「租户隔离」。一个 bug 让 A 公司看到 B 公司的数据,就是安全事故。在 IHUI-AI 这种有 340 张表、144 个迁移、1300+ API 的系统里,我们用 PostgreSQL Row Level Security (RLS) + Drizzle ORM 实现了数据库层的强制隔离——即使应用代码忘了加 WHERE tenant_id = ?,数据库也会拒绝跨租户访问。本文是这套方案的完整工程总结。

一、多租户架构选型

多租户(multi-tenant)架构有三种主流模式,各有取舍。

1.1 三种模式对比

模式	隔离强度	运维成本	扩展性	适合场景
Database-per-tenant	极强	极高	差	金融、医疗合规
Schema-per-tenant	强	高	中	中大型企业客户
Row-level(tenant_id)	中	低	好	SaaS、共享服务

IHUI-AI 选的是 row-level + RLS 强制 模式:

隔离强度够用:不同租户的数据在同一张表,但 RLS 策略保证不可互访
运维成本低:一套 schema,一次迁移,所有租户共享
扩展性好:加租户只是 INSERT 一行,不需要建库建表

1.2 为什么不纯靠应用层

最朴素的方案是每条 SQL 都带 WHERE tenant_id = ?。问题在于:

// 危险:忘了加 WHERE,数据全泄露
await db.select().from(users);

// 危险:JOIN 时漏了 tenant 条件
await db.select()
  .from(orders)
  .leftJoin(items, eq(items.order_id, orders.id)); // items 没 tenant 过滤

// 危险:子查询忘了 tenant
await db.select().from(orders)
  .where(inArray(orders.user_id,
    db.select({id: users.id}).from(users))); // users 没 tenant 过滤

人类总会犯错。340 张表、1300+ API,只要有一个 endpoint 漏了 tenant_id,就是事故。RLS 的价值是把隔离下沉到数据库,应用层犯错也不会泄露。

二、PostgreSQL RLS 基础

2.1 RLS 是什么

Row Level Security 是 PostgreSQL 9.5+ 的内置功能,允许在表上定义「行级访问策略」。开启 RLS 后,所有查询都会被策略过滤,除非显式绕过。

2.2 最小示例

-- 1. 建表(带 tenant_id 列)
CREATE TABLE documents (
    id BIGSERIAL PRIMARY KEY,
    tenant_id VARCHAR(64) NOT NULL,
    title TEXT NOT NULL,
    content TEXT,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- 2. 开启 RLS
ALTER TABLE documents ENABLE ROW LEVEL SECURITY;

-- 3. 创建策略:当前会话变量 app.tenant_id 等于行的 tenant_id 才能读
CREATE POLICY tenant_isolation_select ON documents
    FOR SELECT
    USING (tenant_id = current_setting('app.tenant_id', true));

-- 4. 写策略:只能写自己租户的行
CREATE POLICY tenant_isolation_insert ON documents
    FOR INSERT
    WITH CHECK (tenant_id = current_setting('app.tenant_id', true));

-- 5. 更新/删除策略:只能改自己租户的行
CREATE POLICY tenant_isolation_update ON documents
    FOR UPDATE
    USING (tenant_id = current_setting('app.tenant_id', true))
    WITH CHECK (tenant_id = current_setting('app.tenant_id', true));

CREATE POLICY tenant_isolation_delete ON documents
    FOR DELETE
    USING (tenant_id = current_setting('app.tenant_id', true));

2.3 会话变量怎么设

current_setting('app.tenant_id', true) 读取会话变量。在应用层,每次拿到请求时设置:

-- 设置当前租户(连接级)
SET LOCAL app.tenant_id = 'tenant_abc123';

-- 之后的查询自动被策略过滤
SELECT * FROM documents;  -- 只返回 tenant_id = 'tenant_abc123' 的行

SET LOCAL 只在当前事务内生效,事务结束自动清除,适合请求级隔离。

2.4 超级用户绕过陷阱

重要:PostgreSQL 超级用户(superuser)和表的 OWNER 默认绕过 RLS。生产环境务必:

-- 应用连接用普通角色,不是 superuser
-- 如果必须用 owner,显式强制 RLS
ALTER TABLE documents FORCE ROW LEVEL SECURITY;

FORCE 让 owner 也受策略约束。superuser 仍能绕过(这是 PG 设计),所以应用账号绝不能是 superuser。

三、Drizzle ORM 集成 RLS

Drizzle ORM 0.38 不原生支持 RLS,需要自己写 middleware 在每个请求开始时注入 SET LOCAL。

3.1 连接池 + 请求级事务

// packages/database/src/rls.ts
import { Pool } from "postgres";
import { drizzle, PostgresJsDatabase } from "drizzle-orm/postgres-js";
import * as schema from "./schema";

const pool = new Pool({
  connectionString: process.env.DATABASE_URL!,
  max: 20,
});

export type RlsContext = {
  tenantId: string;
  userId: string;
  roleId: number;
};

/**
 * 在 RLS 上下文中执行函数,自动设置会话变量并在事务结束时清除。
 */
export async function withRls<T>(
  ctx: RlsContext,
  fn: (db: PostgresJsDatabase<typeof schema>) => Promise<T>,
): Promise<T> {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");
    // 注入租户上下文(SET LOCAL 只在事务内有效)
    await client.query(`SET LOCAL app.tenant_id = '${ctx.tenantId}'`);
    await client.query(`SET LOCAL app.user_id = '${ctx.userId}'`);
    await client.query(`SET LOCAL app.role_id = '${ctx.roleId}'`);

    const db = drizzle(client, { schema });
    const result = await fn(db);
    await client.query("COMMIT");
    return result;
  } catch (err) {
    await client.query("ROLLBACK");
    throw err;
  } finally {
    client.release();
  }
}

3.2 Fastify 集成

在 API 层用 preHandler 钩子统一注入:

// apps/api/src/plugins/rls.ts
import { FastifyPluginAsync } from "fastify";
import { withRls, RlsContext } from "@ihui/database";

export const rlsPlugin: FastifyPluginAsync = async (fastify) => {
  fastify.addHook("preHandler", async (request, reply) => {
    const user = request.user; // 来自 auth 插件
    if (!user) return; // 公开路由不注入

    const ctx: RlsContext = {
      tenantId: user.tenantId,
      userId: user.id,
      roleId: user.roleId,
    };

    // 把 rlsContext 挂到 request 上,handler 内用 withRls 执行查询
    request.rlsContext = ctx;
  });
};

// 在路由 handler 里
app.post("/documents", async (request, reply) => {
  return withRls(request.rlsContext, async (db) => {
    // db 已自动注入 tenant 上下文,RLS 策略生效
    const [doc] = await db.insert(schema.documents)
      .values({ title: request.body.title, tenantId: request.rlsContext.tenantId })
      .returning();
    return { code: 0, message: "ok", data: doc };
  });
});

关键点:tenantId 由服务端从认证态注入,绝不信任客户端传的 tenant_id。

四、IHUI-AI 340 张表的 RLS 策略设计

4.1 表分类

不是所有表都需要 RLS。IHUI-AI 把 340 张表分成三类:

类型	数量	是否 RLS	示例
租户表(tenant-owned)	287	是	documents, tasks, messages
共享表(shared)	38	否	models, mcp_servers_registry
全局表(global)	15	否	users, tenants, audit_logs

租户表:每行属于一个租户,必须 tenant_id 列 + RLS 策略
共享表:所有租户共享只读数据(如 LLM 模型清单),不需要 RLS
全局表:系统级数据,通过应用层权限控制(如 admin 后台)

4.2 策略模板

为避免给 287 张表手写策略,我们用脚本批量生成:

// scripts/generate-rls-policies.ts
import { schema } from "../packages/database/src/schema";

const TENANT_TABLES = Object.values(schema)
  .filter((table: any) => table.tenantId !== undefined)
  .map((table: any) => table[Symbol.for("drizzle:Name")]);

function policySQL(tableName: string): string {
  return `
-- RLS for ${tableName}
ALTER TABLE ${tableName} ENABLE ROW LEVEL SECURITY;
ALTER TABLE ${tableName} FORCE ROW LEVEL SECURITY;

CREATE POLICY ${tableName}_select ON ${tableName}
    FOR SELECT USING (tenant_id = current_setting('app.tenant_id', true));

CREATE POLICY ${tableName}_insert ON ${tableName}
    FOR INSERT WITH CHECK (tenant_id = current_setting('app.tenant_id', true));

CREATE POLICY ${tableName}_update ON ${tableName}
    FOR UPDATE USING (tenant_id = current_setting('app.tenant_id', true))
    WITH CHECK (tenant_id = current_setting('app.tenant_id', true));

CREATE POLICY ${tableName}_delete ON ${tableName}
    FOR DELETE USING (tenant_id = current_setting('app.tenant_id', true));
`;
}

console.log(TENANT_TABLES.map(policySQL).join("\n"));

4.3 admin 跨租户查询

admin 后台需要查所有租户的数据。我们用角色判断:

-- admin 角色(role_id >= 1)绕过租户过滤
CREATE POLICY documents_admin_select ON documents
    FOR SELECT
    USING (
        current_setting('app.role_id', true)::int >= 1
        OR tenant_id = current_setting('app.tenant_id', true)
    );

这样 admin 查询时 role_id >= 1,策略放行所有行;普通用户 role_id = 0,只能看自己租户。

五、性能优化

5.1 索引策略

RLS 策略 tenant_id = current_setting(...) 会被加入每条查询的 WHERE 子句。必须在 tenant_id 上建索引:

-- 每张租户表必须有这个索引
CREATE INDEX idx_documents_tenant_id ON documents(tenant_id);

-- 复合索引:常见查询模式(tenant + 时间)
CREATE INDEX idx_documents_tenant_created ON documents(tenant_id, created_at DESC);

-- 复合索引:tenant + 业务字段
CREATE INDEX idx_documents_tenant_status ON documents(tenant_id, status);

实测:无索引时 287 张表的 RLS 查询平均慢 8 倍;加索引后接近无 RLS 的性能。

5.2 查询计划验证

上线前用 EXPLAIN 验证策略命中索引:

SET LOCAL app.tenant_id = 'tenant_test';
EXPLAIN SELECT * FROM documents WHERE status = 'published';

期望看到 Index Scan using idx_documents_tenant_status,而不是 Seq Scan。

5.3 连接池配置

RLS 依赖 SET LOCAL,必须在事务内。连接池要确保:

每个 withRls 调用占一个 client(不共享)
事务结束 client.release(),会话变量自动清除
池大小要够大(每个并发请求占一个连接),IHUI-AI 用 max: 20

六、迁移管理

6.1 drizzle-kit + RLS 同步

drizzle-kit 生成 schema 迁移时,不会自动加 RLS 策略。我们的工作流:

pnpm --filter @ihui/database db:generate 生成 SQL 迁移
手动在迁移文件末尾追加 RLS 策略(用脚本生成)
pnpm --filter @ihui/database db:migrate 执行

// packages/database/src/migrations/0089_add_notifications.ts
import { sql } from "drizzle-orm";
import { pgTable, varchar, text, timestamp } from "drizzle-orm/pg-core";

export const notifications = pgTable("notifications", {
  id: varchar("id").primaryKey(),
  tenantId: varchar("tenant_id").notNull(),  // 关键:必须有
  userId: varchar("user_id").notNull(),
  message: text("message").notNull(),
  createdAt: timestamp("created_at").defaultNow().notNull(),
});

// 迁移 SQL(自动生成 + 手动追加 RLS)
export const migrationSQL = sql.raw(`
CREATE TABLE "notifications" (
    "id" varchar PRIMARY KEY NOT NULL,
    "tenant_id" varchar NOT NULL,
    "user_id" varchar NOT NULL,
    "message" text NOT NULL,
    "created_at" timestamp DEFAULT now() NOT NULL
);
CREATE INDEX idx_notifications_tenant_id ON "notifications"("tenant_id");

-- RLS 策略(脚本生成,手动粘贴)
ALTER TABLE "notifications" ENABLE ROW LEVEL SECURITY;
ALTER TABLE "notifications" FORCE ROW LEVEL SECURITY;
CREATE POLICY notifications_select ON "notifications"
    FOR SELECT USING (tenant_id = current_setting('app.tenant_id', true));
CREATE POLICY notifications_insert ON "notifications"
    FOR INSERT WITH CHECK (tenant_id = current_setting('app.tenant_id', true));
CREATE POLICY notifications_update ON "notifications"
    FOR UPDATE USING (tenant_id = current_setting('app.tenant_id', true))
    WITH CHECK (tenant_id = current_setting('app.tenant_id', true));
CREATE POLICY notifications_delete ON "notifications"
    FOR DELETE USING (tenant_id = current_setting('app.tenant_id', true));
`);

6.2 守门脚本:防止漏加 RLS

我们写了 scripts/check-rls-policies.mjs 守门,在 pre-commit 阶段检查:

所有 schema 里带 tenantId 的表,迁移文件里必须有对应的 ENABLE ROW LEVEL SECURITY
缺失 → 阻塞 commit

// scripts/check-rls-policies.mjs(简化版)
import { glob } from "glob";
import { readFileSync } from "fs";

const migrationFiles = await glob("packages/database/src/migrations/*.ts");
const schemaFiles = await glob("packages/database/src/schema/*.ts");

// 提取所有有 tenantId 的表名
const tenantTables = extractTenantTables(schemaFiles);
const errors = [];

for (const table of tenantTables) {
  // 检查至少有一个迁移文件包含该表的 RLS 策略
  const hasRLS = migrationFiles.some(f =>
    readFileSync(f, "utf-8").includes(`ENABLE ROW LEVEL SECURITY`) &&
    readFileSync(f, "utf-8").includes(table)
  );
  if (!hasRLS) errors.push(`Table ${table} missing RLS policy`);
}

if (errors.length) {
  console.error("RLS policy check failed:");
  errors.forEach(e => console.error("  -", e));
  process.exit(1);
}

七、测试策略

7.1 三类测试

// apps/api/tests/rls.test.ts
import { describe, it, expect } from "vitest";
import { withRls, pool } from "@ihui/database";

describe("RLS 隔离测试", () => {
  it("租户 A 查不到租户 B 的数据", async () => {
    // 用 tenantA 写一条
    await withRls({ tenantId: "tenantA", userId: "u1", roleId: 0 }, async (db) => {
      await db.insert(schema.documents).values({ id: "d1", tenantId: "tenantA", title: "A" });
    });

    // 用 tenantB 查,应该查不到
    await withRls({ tenantId: "tenantB", userId: "u2", roleId: 0 }, async (db) => {
      const docs = await db.select().from(schema.documents);
      expect(docs.find(d => d.id === "d1")).toBeUndefined();
    });
  });

  it("普通用户不能写入其他租户的数据", async () => {
    await expect(
      withRls({ tenantId: "tenantA", userId: "u1", roleId: 0 }, async (db) => {
        // 故意写 tenantB 的数据,应该被 RLS 拒绝
        await db.insert(schema.documents).values({ id: "d2", tenantId: "tenantB", title: "X" });
      }),
    ).rejects.toThrow(); // WITH CHECK 失败
  });

  it("admin 可以跨租户查询", async () => {
    await withRls({ tenantId: "tenantA", userId: "admin", roleId: 1 }, async (db) => {
      const docs = await db.select().from(schema.documents);
      // admin 能看到所有租户
      expect(docs.length).toBeGreaterThan(0);
    });
  });
});

7.2 权限边界测试

每张租户表至少测 4 个场景:

SELECT:租户 A 查不到租户 B
INSERT:租户 A 不能写租户 B 的 tenant_id
UPDATE:租户 A 不能改租户 B 的行
DELETE:租户 A 不能删租户 B 的行

340 张表没法逐个手写,我们用参数化测试批量跑:

// 参数化:对所有租户表跑同一组测试
for (const table of TENANT_TABLES) {
  describe(`${table} RLS`, () => {
    it("租户隔离 SELECT", async () => testTenantIsolation(table, "select"));
    it("租户隔离 INSERT", async () => testTenantIsolation(table, "insert"));
    it("租户隔离 UPDATE", async () => testTenantIsolation(table, "update"));
    it("租户隔离 DELETE", async () => testTenantIsolation(table, "delete"));
  });
}

八、常见陷阱

8.1 超级用户绕过

如前所述,superuser 和 owner(除非 FORCE)绕过 RLS。生产环境:

应用账号用普通 role,CREATE ROLE ihui_app LOGIN PASSWORD '...'
每张表 FORCE ROW LEVEL SECURITY
绝不把 DATABASE_URL 配成 superuser

8.2 外键级联

外键的 ON DELETE CASCADE 不受 RLS 约束。如果租户 A 的订单引用了租户 B 的产品(本身不该发生),删产品时会级联删订单,绕过 RLS。

解法:外键必须包含 tenant_id 联合:

-- 错误:外键不带 tenant,可能跨租户引用
FOREIGN KEY (product_id) REFERENCES products(id)

-- 正确:联合外键,确保同租户
FOREIGN KEY (tenant_id, product_id) REFERENCES products(tenant_id, id)

8.3 CTE 和子查询

RLS 对 CTE(WITH 子句)和子查询都生效,这是好事。但要注意性能:

-- RLS 会自动给 orders 和 users 都加 tenant 过滤
SELECT * FROM orders
WHERE user_id IN (SELECT id FROM users WHERE status = 'active');
-- 实际执行:orders.tenant_id = X AND users.tenant_id = X

如果 CTE 里用了 MATERIALIZED,RLS 策略可能不被优化器下推,导致全表扫。生产环境慎用 MATERIALIZED。

8.4 COPY 绕过

COPY 命令默认绕过 RLS(除非 FORCE)。备份/导入数据时要注意,别用应用账号跑 COPY。

九、监控与审计

9.1 RLS 命中统计

-- 查询哪些表的 RLS 策略被频繁命中
SELECT relname, n_tup_ins, n_tup_upd, n_tup_del, n_tup_sel
FROM pg_stat_user_tables
WHERE relname IN (SELECT tablename FROM pg_policies);

9.2 审计日志

关键操作(跨租户 admin 查询、批量导出)记录到 audit_logs 表:

CREATE TABLE audit_logs (
    id BIGSERIAL PRIMARY KEY,
    admin_id VARCHAR(64) NOT NULL,
    action VARCHAR(32) NOT NULL,  -- cross_tenant_select | export | delete
    target_tenant VARCHAR(64),    -- 被查询的目标租户
    details JSONB,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

十、参考资料

PostgreSQL RLS 官方文档
Drizzle ORM 文档
Supabase RLS 实践
IHUI-AI 源码:packages/database/src/rls.ts + scripts/check-rls-policies.mjs

总结

在 340 张表的生产 SaaS 里,纯靠应用层 WHERE tenant_id = ? 是不可靠的——人总会犯错。PostgreSQL RLS 把隔离下沉到数据库,即使应用代码漏了过滤,数据库也会拒绝跨租户访问,这是深度防御的典范。

IHUI-AI 的实践要点:

row-level + RLS 强制:287 张租户表全开 RLS + FORCE
Drizzle middleware 注入:withRls 在事务内 SET LOCAL 会话变量
脚本批量生成策略:避免手写 287 份 SQL
守门脚本:pre-commit 检查新增表必须有 RLS
参数化测试:4 场景 × 287 表自动验证隔离

下一篇讲 LangGraph Agent 编排模式,看多 agent 如何协作完成复杂任务。

This article was originally published on the IHUI AI Blog. Follow us on GitHub for more AI engineering content.

MCP 协议实现指南:从零构建生产级 AI 工具生态

李春川 — Tue, 28 Jul 2026 05:38:23 +0000

MCP 协议实现指南:从零构建生产级 AI 工具生态

当 LLM 不再只是「聊天机器人」,而是要查数据库、发邮件、调内部 API 时,工具调用就成了 AI 应用的命脉。Anthropic 在 2024 年开源的 Model Context Protocol (MCP),正在成为 AI 工具生态的「USB-C 接口」。本文是 IHUI-AI 在生产环境落地 MCP 的完整工程总结。

一、MCP 协议是什么

Model Context Protocol(简称 MCP)是 Anthropic 于 2024 年 11 月开源的开放协议,目标是标准化 AI 应用与外部工具/数据源之间的连接方式。

在 MCP 出现之前,每个 AI 应用都要为每个工具写一份适配代码:

┌─────────┐    自定义协议 A    ┌──────────┐
│  App 1  │ ──────────────────▶│ Tool GPT │
└─────────┘                    └──────────┘
┌─────────┐    自定义协议 B    ┌──────────┐
│  App 2  │ ──────────────────▶│ Tool GPT │
└─────────┘                    └──────────┘

M 个应用 × N 个工具 = M×N 份适配代码,这是典型的「集成地狱」。

MCP 的解法是引入一个标准协议层:

┌─────────┐                  ┌──────────┐
│  App 1  │ ──┐          ┌──▶│ Tool A   │
└─────────┘   │          │   └──────────┘
┌─────────┐   │  ┌───────┐   ┌──────────┐
│  App 2  │───┼─▶│  MCP  │──▶│ Tool B   │
└─────────┘   │  └───────┘   └──────────┘
┌─────────┐   │          │   ┌──────────┐
│  App 3  │ ──┘          └──▶│ Tool C   │
└─────────┘                  └──────────┘

M + N 份适配代码,复杂度从 O(M×N) 降到 O(M+N)。

1.1 MCP 的三大原语

MCP 定义了三种核心原语(primitives):

Tools(工具):可被 LLM 调用的函数,类似 function calling,但有标准 schema
Resources(资源):可被读取的数据源,如文件、数据库记录、API 响应
Prompts(提示模板):可复用的 prompt 模板,支持参数化

1.2 传输层

MCP 支持两种传输方式:

stdio:本地进程通信,适合 CLI / 桌面端
HTTP + SSE:远程通信,适合 Web / 云端服务

二、为什么 MCP 比传统 function calling 更适合生产环境

很多人第一反应是:「这不就是 function calling 换个名字吗?」差远了。

2.1 function calling 的痛点

OpenAI 在 2023 年推出 function calling 时,设计目标是「让 LLM 输出结构化 JSON」。它解决的是单次调用问题,不是工具生态问题:

维度	function calling	MCP
工具发现	硬编码在 prompt 里	运行时动态发现(list_tools)
工具复用	每个 app 重写一遍	一次实现,处处可用
跨厂商	OpenAI/Anthropic 格式不同	协议层统一
状态管理	无状态,每次重传	有 session,支持长连接
权限模型	无	内置 capability negotiation
流式响应	不支持	支持(SSE)

2.2 生产场景的真实需求

在 IHUI-AI 这种多 agent 平台里,我们需要:

动态接入新工具:用户上传一个 MCP server 配置,系统自动发现工具,无需重启
多 agent 共享工具:聊天 agent 和任务 agent 都能调用同一个 search_knowledge_base
权限隔离:免费用户只能用只读工具,付费用户能用写工具
审计日志:每次工具调用都要记录 who/when/what/result

这些 function calling 都做不到,而 MCP 的协议设计天然支持。

三、MCP Server 实现详解

MCP Server 是工具的提供方。下面用 Python 和 TypeScript 各实现一个最小可用的 MCP server。

3.1 Python 实现(FastMCP)

Anthropic 官方提供了 fastmcp 库,把 MCP server 实现简化到装饰器级别:

# apps/ai-service/app/mcp/servers/knowledge_base.py
from fastmcp import FastMCP
from typing import Optional
from app.core.vector_store import pgvector_search
from app.core.auth import verify_tenant

mcp = FastMCP("knowledge-base")

@mcp.tool()
async def search_kb(
    query: str,
    top_k: int = 5,
    tenant_id: str = "",
    knowledge_base_id: Optional[str] = None,
) -> dict:
    """在知识库中语义检索相关文档。

    Args:
        query: 用户的查询文本
        top_k: 返回的最大文档数量,默认 5
        tenant_id: 租户 ID,用于多租户隔离
        knowledge_base_id: 可选,限定在某个知识库内检索

    Returns:
        包含 documents 和 scores 的字典
    """
    if not tenant_id:
        return {"error": "tenant_id is required"}

    # 多租户隔离:确保只能查到本租户的数据
    await verify_tenant(tenant_id)

    results = await pgvector_search(
        query=query,
        top_k=top_k,
        tenant_id=tenant_id,
        kb_id=knowledge_base_id,
    )
    return {
        "documents": [r.document for r in results],
        "scores": [r.score for r in results],
    }

@mcp.resource("kb://list/{tenant_id}")
async def list_knowledge_bases(tenant_id: str) -> dict:
    """列出租户下所有知识库。"""
    bases = await get_kbs_by_tenant(tenant_id)
    return {"knowledge_bases": [b.to_dict() for b in bases]}

if __name__ == "__main__":
    mcp.run(transport="stdio")

关键点:

@mcp.tool() 装饰器自动从函数签名生成 JSON Schema
docstring 会被解析成工具描述,LLM 靠它判断何时调用
tenant_id 强制要求,实现多租户隔离
@mcp.resource() 暴露只读数据源,用 URI 模式寻址

3.2 TypeScript 实现(@modelcontextprotocol/sdk)

对于 Node.js 生态,官方提供 @modelcontextprotocol/sdk:

// apps/api/src/mcp/servers/file-ops.ts
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";

const server = new Server(
  { name: "file-ops", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

// 工具清单:运行时被 client 发现
server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [
    {
      name: "read_file",
      description: "读取本地文件内容(限制在允许目录内)",
      inputSchema: {
        type: "object",
        properties: {
          path: { type: "string", description: "文件绝对路径" },
          max_bytes: { type: "number", description: "最大读取字节数", default: 1048576 },
        },
        required: ["path"],
      },
    },
    {
      name: "list_dir",
      description: "列出目录内容",
      inputSchema: {
        type: "object",
        properties: { path: { type: "string" } },
        required: ["path"],
      },
    },
  ],
}));

// 工具调用分发
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  switch (name) {
    case "read_file": {
      const path = sanitizePath(args?.path as string); // 防 path traversal
      const maxBytes = (args?.max_bytes as number) ?? 1048576;
      const content = await readFileTruncated(path, maxBytes);
      return { content: [{ type: "text", text: content }] };
    }
    case "list_dir": {
      const path = sanitizePath(args?.path as string);
      const entries = await listDir(path);
      return { content: [{ type: "text", text: JSON.stringify(entries) }] };
    }
    default:
      throw new Error(`Unknown tool: ${name}`);
  }
});

const transport = new StdioServerTransport();
await server.connect(transport);

3.3 关键设计原则

实现生产级 MCP server 时,务必遵守:

工具描述要写给 LLM 看:不是给人看的 API 文档,而是让 LLM 能判断「这个工具该不该用」
参数 schema 要严格:用 JSON Schema 的 required / enum / pattern,减少 LLM 瞎猜
错误信息要可读:返回 {"error": "tenant_id missing"} 而不是抛 500
幂等设计:同一个工具调用,带相同参数应该返回相同结果(便于重试)

四、MCP Client 集成模式

MCP Client 是工具的调用方,通常嵌在 AI agent 内部。有三种主流集成模式。

4.1 模式一:直接调用(适合简单场景)

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def call_kb_tool(query: str, tenant_id: str):
    server_params = StdioServerParameters(
        command="python",
        args=["-m", "app.mcp.servers.knowledge_base"],
    )
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            result = await session.call_tool(
                "search_kb",
                {"query": query, "tenant_id": tenant_id, "top_k": 5},
            )
            return result

缺点:每次调用都新建进程,开销大。适合开发调试,不适合生产。

4.2 模式二:连接池(适合生产)

IHUI-AI 在 apps/ai-service/app/mcp/pool.py 实现了连接池,复用 MCP session:

# apps/ai-service/app/mcp/pool.py
from contextlib import asynccontextmanager
from mcp import ClientSession
from mcp.client.sse import sse_client
from collections import defaultdict
import asyncio

class MCPSessionPool:
    """MCP session 连接池,按 server_url 复用长连接。"""

    def __init__(self, max_size_per_server: int = 8):
        self._pools: dict[str, asyncio.Queue] = defaultdict(lambda: asyncio.Queue(maxsize=max_size_per_server))
        self._lock = asyncio.Lock()

    @asynccontextmanager
    async def acquire(self, server_url: str):
        async with self._lock:
            pool = self._pools[server_url]
            if pool.empty() and pool.qsize() < pool.maxsize:
                session = await self._create_session(server_url)
            else:
                session = await pool.get()

        try:
            yield session
        finally:
            await pool.put(session)

    async def _create_session(self, server_url: str) -> ClientSession:
        read, write = await sse_client(server_url)
        session = ClientSession(read, write)
        await session.initialize()
        return session

# 全局单例
mcp_pool = MCPSessionPool(max_size_per_server=8)

调用方:

async with mcp_pool.acquire("http://localhost:8820/sse") as session:
    result = await session.call_tool("search_kb", {"query": q, "tenant_id": tid})

4.3 模式三:与 LangGraph 集成(推荐)

LangGraph 是 LangChain 的下一代 agent 框架,天然支持把 MCP tools 绑定到 agent 节点:

# apps/ai-service/app/graphs/chat_graph.py
from langgraph.prebuilt import create_react_agent
from langchain_mcp_adapters.client import MultiServerMCPClient

async def build_chat_agent(tenant_id: str, user_role: str):
    client = MultiServerMCPClient({
        "knowledge_base": {
            "url": "http://localhost:8820/sse",
            "transport": "sse",
        },
        "file_ops": {
            "url": "http://localhost:8821/sse",
            "transport": "sse",
        },
    })

    # 拉取所有 server 的工具清单
    all_tools = await client.get_tools()

    # 按用户角色过滤工具(权限隔离)
    allowed = filter_tools_by_role(all_tools, user_role)

    # 注入 tenant_id 到每个工具的默认参数
    for tool in allowed:
        tool = inject_default_arg(tool, "tenant_id", tenant_id)

    agent = create_react_agent(
        model="openai/gpt-4o",
        tools=allowed,
        state_schema=ChatState,
    )
    return agent

这种模式的好处:

工具发现自动化:server 一更新,client 自动拉到新工具
权限隔离在 agent 层:free user 的 agent 根本看不到付费工具
状态传递:tenant_id 通过 inject_default_arg 注入,LLM 不需要每次传

五、工具发现 + 权限 + 生命周期管理

5.1 工具发现流程

┌──────────────┐    initialize     ┌──────────────┐
│  MCP Client  │ ─────────────────▶│  MCP Server  │
└──────────────┘                   └──────────────┘
       │                                  │
       │  list_tools()                    │
       │ ────────────────────────────────▶│
       │                                  │ 返回工具 schema 列表
       │◀──────────────────────────────── │
       │                                  │
       │  call_tool(name, args)           │
       │ ────────────────────────────────▶│
       │                                  │ 执行并返回结果
       │◀──────────────────────────────── │

5.2 权限模型

IHUI-AI 实现了三层权限:

# apps/ai-service/app/mcp/permissions.py
TOOL_PERMISSIONS = {
    # 工具名: (最小角色, 是否需要二次确认)
    "search_kb":            ("user", False),
    "create_kb":            ("user", True),   # 写操作需确认
    "delete_kb":            ("admin", True),
    "read_file":            ("user", False),
    "write_file":           ("admin", True),
    "execute_sql":          ("admin", True),
    "send_email":           ("user", True),
}

def check_permission(tool_name: str, user_role: str) -> bool:
    required_role, _ = TOOL_PERMISSIONS.get(tool_name, ("admin", True))
    role_level = {"guest": 0, "user": 1, "admin": 2}
    return role_level.get(user_role, 0) >= role_level[required_role]

5.3 生命周期

MCP server 的生命周期由 client 管理:

启动:client 第一次连接时,server 进程拉起(stdio)或连接建立(SSE)
健康检查:client 定期 ping,超时 3 次判定失联
优雅关闭:client 发 shutdown 通知,server 清理资源后退出
强制回收:超过最大空闲时间(默认 30 分钟)的 session 被回收

六、IHUI-AI 中的 MCP 实战

IHUI-AI 的 MCP 实现位于 apps/ai-service/app/mcp/,目录结构:

apps/ai-service/app/mcp/
├── __init__.py
├── pool.py              # 连接池
├── permissions.py       # 权限模型
├── registry.py          # server 注册表(动态加载)
├── audit.py             # 审计日志
├── servers/             # 内置 MCP server 实现
│   ├── knowledge_base.py
│   ├── file_ops.py
│   ├── sql_query.py
│   └── web_search.py
└── client.py            # 统一 client 入口

6.1 动态注册表

用户可以通过 Web UI 上传自定义 MCP server 配置,系统动态加载:

# apps/ai-service/app/mcp/registry.py
from pydantic import BaseModel
from typing import Optional

class MCPServerConfig(BaseModel):
    name: str
    transport: str  # "stdio" | "sse"
    command: Optional[str] = None      # stdio 模式
    args: list[str] = []               # stdio 模式
    url: Optional[str] = None          # sse 模式
    enabled: bool = True
    required_role: str = "user"

class MCPRegistry:
    """MCP server 注册表,从数据库加载配置。"""

    async def load_for_tenant(self, tenant_id: str) -> list[MCPServerConfig]:
        rows = await db.execute(
            "SELECT * FROM mcp_servers WHERE tenant_id = $1 AND enabled = TRUE",
            tenant_id,
        )
        return [MCPServerConfig(**row) for row in rows]

    async def discover_tools(self, configs: list[MCPServerConfig]) -> list[Tool]:
        tools = []
        for cfg in configs:
            async with mcp_pool.acquire(cfg.url or cfg.command) as session:
                server_tools = await session.list_tools()
                tools.extend(server_tools)
        return tools

6.2 审计日志

每次工具调用都记录到 mcp_tool_calls 表:

CREATE TABLE mcp_tool_calls (
    id BIGSERIAL PRIMARY KEY,
    tenant_id VARCHAR(64) NOT NULL,
    user_id VARCHAR(64) NOT NULL,
    server_name VARCHAR(128) NOT NULL,
    tool_name VARCHAR(128) NOT NULL,
    arguments JSONB NOT NULL,
    result JSONB,
    status VARCHAR(16) NOT NULL,  -- success | error | timeout
    duration_ms INT NOT NULL,
    tokens_used INT DEFAULT 0,
    cost_usd DECIMAL(10,6) DEFAULT 0,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_mcp_calls_tenant_time ON mcp_tool_calls(tenant_id, created_at DESC);
CREATE INDEX idx_mcp_calls_tool ON mcp_tool_calls(tenant_id, tool_name);

admin 后台可以按租户/工具/时间维度查询调用统计,做成本归因。

七、与 A2A 协议的协同

MCP 解决的是「agent ↔ tool」的通信,A2A(Agent-to-Agent)解决的是「agent ↔ agent」的通信。两者互补:

┌─────────────────────────────────────────────────┐
│              Agent Orchestration Layer           │
│         (LangGraph / A2A Protocol)               │
└──────┬───────────────────────┬──────────────────┘
       │ A2A                   │ A2A
       ▼                       ▼
┌──────────────┐         ┌──────────────┐
│ Researcher   │ ──MCP──▶│ Tools        │
│ Agent        │         │ (KB, Web,    │
│              │ ──MCP──▶│  SQL, File)  │
└──────────────┘         └──────────────┘
       │ A2A
       ▼
┌──────────────┐
│ Writer Agent │ ──MCP──▶│ Tools (Doc)  │
└──────────────┘

MCP:agent 调用无状态工具(查数据库、读文件)
A2A:agent 之间协商任务(研究 agent 把结果交给写作 agent)

IHUI-AI 的多 agent 编排里,researcher agent 通过 MCP 调用 search_kb / web_search,然后通过 A2A 把结果传给 writer agent,writer 再通过 MCP 调用 create_doc 工具产出最终文档。

八、性能优化

8.1 连接池

生产环境务必用连接池(见 4.2),避免每次调用都新建 TCP 连接 + MCP handshake。IHUI-AI 的实测数据:

模式	单次调用延迟	QPS
每次新建连接	180ms	12
连接池(8 conn)	23ms	320

8.2 并发控制

多 agent 并行调用同一 server 时,要做并发限制,避免打爆下游:

import asyncio

class ToolCallLimiter:
    def __init__(self, max_concurrent: int = 10):
        self._semaphore = asyncio.Semaphore(max_concurrent)

    async def call(self, session, tool_name, args):
        async with self._semaphore:
            return await session.call_tool(tool_name, args)

8.3 超时与重试

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type(TimeoutError),
)
async def call_with_retry(session, tool_name, args, timeout=30):
    return await asyncio.wait_for(
        session.call_tool(tool_name, args),
        timeout=timeout,
    )

注意:只对幂等工具重试。create_kb 这种写工具失败后重试会创建重复记录,要用 idempotency key。

8.4 结果缓存

对于只读且结果稳定的工具(如 search_kb 同 query 在 5 分钟内),可以缓存:

from functools import lru_cache
import hashlib
import json

def cache_key(tool_name, args):
    raw = json.dumps({"tool": tool_name, "args": args}, sort_keys=True)
    return hashlib.sha256(raw.encode()).hexdigest()

# 用 Redis 做分布式缓存,ttl 300 秒
async def cached_call(session, tool_name, args, ttl=300):
    key = f"mcp:cache:{cache_key(tool_name, args)}"
    cached = await redis.get(key)
    if cached:
        return json.loads(cached)
    result = await session.call_tool(tool_name, args)
    await redis.setex(key, ttl, json.dumps(result))
    return result

九、安全考量

9.1 工具白名单

永远不要让 LLM 自由调用任意工具。在 agent 层做白名单过滤:

ALLOWED_TOOLS_PER_AGENT = {
    "chat_agent":       ["search_kb", "web_search"],
    "task_agent":       ["search_kb", "create_task", "update_task"],
    "admin_agent":      ["*"],  # admin 才能用所有工具
}

9.2 参数校验

LLM 可能生成畸形参数(如 path 里带 ../)。在工具实现层做 sanitize:

import re
from pathlib import Path

ALLOWED_ROOT = Path("/data/sandbox")

def sanitize_path(user_path: str) -> Path:
    p = (ALLOWED_ROOT / user_path).resolve()
    if not str(p).startswith(str(ALLOWED_ROOT)):
        raise ValueError(f"Path escapes sandbox: {user_path}")
    return p

9.3 审计日志

见 6.2,所有调用记录到 mcp_tool_calls 表,支持事后追溯。敏感字段(如 API key 参数)在写入前做脱敏。

9.4 速率限制

per-tenant + per-tool 的速率限制,防滥用:

from slowapi import Limiter

limiter = Limiter(key_func=lambda req: req.state.tenant_id)

@app.post("/mcp/call/{tool_name}")
@limiter.limit("60/minute")  # 每租户每分钟 60 次
async def call_tool(tool_name: str, ...):
    ...

十、调试技巧

10.1 MCP Inspector

Anthropic 官方提供 @modelcontextprotocol/inspector,可视化调试 MCP server:

npx @modelcontextprotocol/inspector python -m app.mcp.servers.knowledge_base

打开浏览器可以看到:

所有工具的 schema
手动填参数测试调用
查看原始 JSON-RPC 通信

10.2 日志埋点

在 client 层记录每次调用的入参/出参/耗时:

import structlog
logger = structlog.get_logger()

async def logged_call(session, tool_name, args):
    logger.info("mcp_call_start", tool=tool_name, args=args)
    start = time.monotonic()
    try:
        result = await session.call_tool(tool_name, args)
        logger.info("mcp_call_success",
                    tool=tool_name, duration_ms=int((time.monotonic()-start)*1000))
        return result
    except Exception as e:
        logger.error("mcp_call_error", tool=tool_name, error=str(e))
        raise

10.3 断点调试

stdio 模式的 MCP server 可以直接在 IDE 里断点调试,因为它是子进程。SSE 模式可以单独启动 server,然后用 client 连上去调试。

十一、路线图:MCP 1.0 → 2.0 演进

MCP 协议仍在快速演进。已知的发展方向:

Streamable HTTP:2.0 计划用单一 HTTP transport 替代 stdio + SSE 双轨,简化部署
工具版本化:支持 search_kb@v2,平滑升级不破坏旧 client
Resource 订阅:client 可以订阅 resource 变更(server push 而非轮询)
A2A 标准化:agent 间通信协议层标准化,与 MCP 形成完整生态
官方注册表:类似 npm registry 的 MCP server 市场,一键安装

IHUI-AI 会持续跟进协议演进,目前在 1.x 稳定版上生产运行。

十二、参考资料

总结

MCP 不只是「function calling 的标准化」,它定义了一套完整的工具生态协议:发现、调用、权限、生命周期、审计。在 IHUI-AI 的实践中,MCP 让我们能:

动态接入工具:用户上传配置即可,无需重启服务
多 agent 共享:聊天 agent 和任务 agent 复用同一套工具实现
权限隔离:按角色过滤工具,按租户隔离数据
成本归因:每次调用记录 token 和成本,精准计费

如果你在做 AI agent 平台,MCP 是绕不开的基础设施。建议从 Python FastMCP 起步,先实现一个 search_kb 工具体验全流程,再逐步迁移到生产架构(连接池 + 权限 + 审计)。

下一篇我们会讲 PostgreSQL 多租户 RLS 实践,看 MCP 工具背后的数据层如何做租户隔离。

This article was originally published on the IHUI AI Blog. Follow us on GitHub for more AI engineering content.

开源项目的 7 种 SaaS 变现模式:从 Open Core 到 API 计费

李春川 — Tue, 28 Jul 2026 05:38:18 +0000

开源项目的 7 种 SaaS 变现模式:从 Open Core 到 API 计费

「开源」和「赚钱」并不矛盾。GitLab、HashiCorp、Supabase、Vercel 都从开源项目起步,各自走出不同的变现路径。本文梳理 7 种被市场验证过的开源 SaaS 变现模式,以及 IHUI-AI 是如何组合它们的。

一、开源变现的核心命题

开源不是「免费送代码」,而是「用免费换信任,用信任换付费意愿」。它的商业逻辑建立在三个事实之上:

用户能看代码 → 信任度高 → 转化率比闭源高 3-5 倍。
社区传播 → 获客成本接近零。
客户自己能部署 → 不付钱的用户不消耗你的运维成本。

难点在于:如何把「能用」和「想付费」之间画一条不越界的线。

二、7 种 SaaS 变现模式

模式 1:Open Core(开放核心 + 付费企业功能)

做法:核心开源(Apache 2.0 / MIT),企业必需功能(SSO、审计、RBAC、合规)闭源收费。

案例:GitLab CE / EE、Supabase、Sentry。

适用:企业版有明确合规需求的功能。

# IHUI-AI 的 Open Core 边界
open_source:
  - 8 端代码(Web/API/AI-service/CLI/...)
  - 基础 Agent 编排
  - MCP 客户端/服务端
  - RAG 知识库
enterprise:
  - 多租户 SSO(SAML / OIDC)
  - 审计日志合规导出
  - 高级 RBAC 与权限审批流
  - 私有化部署支持

模式 2:托管 SaaS(Managed Hosting)

做法:用户可以自己部署,但「让官方帮你跑」要付费。

案例:Vercel(Next.js)、Supabase Cloud、Railway。

适用:用户怕运维、想要 SLA、不想自己升级。

定价参考:成本 = 基础设施成本 × 1.5-3,订阅起步价 $19-49/月。

模式 3:企业版许可(Enterprise License)

做法:同一份代码,企业版加授权 + 高级特性 + 法律保障。

案例:Mattermost Enterprise、HashiCorp Vault Enterprise。

适合 To B 项目:企业愿意为「合规背书」「 indemnification 条款」付费。

模式 4:API 计费(Usage-based API)

做法:开源 SDK / 服务,按调用量计费。

案例:OpenAI API、Anthropic API、Replicate。

适用:计算密集型服务(LLM 推理、向量检索、视频转码)。

// IHUI-AI 的 API 计费结构示例
const pricing = {
  chat: { per_1k_tokens: 0.002 },          // 对话
  embedding: { per_1k_tokens: 0.0001 },     // 向量嵌入
  rag_query: { per_query: 0.01 },           // RAG 检索
  agent_run: { per_run: 0.05 },             // Agent 单次执行
  mcp_tool_call: { per_call: 0.001 },       // MCP 工具调用
};

按调用计费的关键设计:给用户「预算上限」(max_budget_per_user),防止账单失控引发信任危机。

模式 5:市场抽成(Marketplace Take Rate)

做法:开放第三方扩展市场,从交易中抽成。

案例:Shopify App Store(10-20%)、VS Code Marketplace、Figma Community。

适用:有第三方创作者生态的项目。

抽成比例参考:10-30%,根据提供的服务(支付 / 托管 / 推广)而定。IHUI-AI Agent 市场对创作者采用 30% 抽成(含支付手续费 + 平台推广 + 质量评估成本)。

模式 6:咨询与定制开发

做法:开源代码免费,部署 / 集成 / 定制收费。

案例:大多数 Apache 项目(Elastic、Confluent for Kafka)。

适用:To B 重型项目,客户需要现场实施。

定价:人天 $1000-3000,合同通常 $50k 起步。

模式 7:认证与培训(Certification & Training)

做法:开源技术免费,但「官方认证」要付费。

案例:CNCF(Kubernetes 认证)、Linux Foundation、MongoDB University。

适合:技术栈复杂、需要人才生态的项目。

定价:认证考试 $200-600,企业培训 $5k-50k/期。

三、IHUI-AI 的 7 大收入流实践

IHUI-AI 不是一个模式,而是组合多个:

模式	IHUI-AI 实现	占比预估
Open Core	核心 8 端开源,企业 RBAC/SSO 闭源	转化入口
托管 SaaS	ihui.ai 一键开通,免运维	40%
企业版	私有部署 + 审计 + 合规	25%
API 计费	按调用计费(token + RAG + Agent)	20%
市场抽成	Agent 市场创作者分成(30%)	10%
咨询定制	大客户私有化集成	4%
认证培训	IHUI Agent 工程师认证	1%

四、变现设计的三个红线

不要把核心功能做成付费:用户能从代码看到「关键功能被锁」,信任崩塌。Open Core 的边界必须是「企业才需要」的功能。
不要走 AGPL 强传染:除非有强力法务,AGPL 会让企业客户绕道走。Apache 2.0 + 商业附加条款更友好。
不要让免费版有容量陷阱:用户用着用着突然限制 100 条数据,体验极差。要么明确免费额度,要么不限容量但限功能。

五、变现启动的三个时机

时机	信号	该做什么
GitHub Star 1k+	社区认可	上线 SaaS 托管
企业用户问「能不能签合同」	付费意愿	推出企业版
第三方问「能不能做插件」	生态成型	开放市场

IHUI-AI 当前处于第一阶段,核心目标是把 SaaS 托管与企业版做扎实。

六、给开源作者的建议

第一天就想好商业模式:不是「先开源,赚钱以后再说」,而是「开源策略服务商业模式」。
企业版功能要早埋:不要等到要变现才匆忙加 RBAC,那时架构已经定型。
License 选对:Apache 2.0 最宽松,适合做 SaaS;GPL 适合强保护;AGPL 谨慎用。
社区贡献 ≠ 商业贡献:不要被「活跃 contributor」绑架,付费客户才是生命线。

相关链接

项目仓库(Apache 2.0):https://github.com/IHUI-INF-AI/IHUI-AI
官网(SaaS 托管入口):https://ihui.ai

如果你也是开源作者或正在考虑商业化,欢迎到 GitHub 给 IHUI-AI 点 Star ⭐,也欢迎来官网看我们如何把开源与商业结合。

SEO 关键词:开源变现、SaaS 订阅、open core、商业化、Apache 2.0 商用、API 计费、企业版、开源商业模式

This article was originally published on the IHUI AI Blog. Follow us on GitHub for more AI engineering content.

AI Agent 市场设计:让 Agent 像 App 一样被交易与编排

李春川 — Tue, 28 Jul 2026 05:38:15 +0000

AI Agent 市场设计:让 Agent 像 App 一样被交易与编排

App Store 把「软件」变成了可被一键购买、安装、评分的商品,AI 时代对应的实体是「Agent」。一个 Agent = 一段可被复用的提示词 + 工具集 + 知识库 + 模型路由配置。本文讲清楚一个 Agent 市场需要哪些核心机制,以及 IHUI-AI 的实现路径。

一、Agent 市场的产品定义

什么是「可上架的 Agent」

不是所有对话 prompt 都能成为商品。一个可上架的 Agent 必须满足:

可独立运行:用户购买后能立刻用,不需要再写代码。
可复用:不同用户用同一个 Agent 都能得到稳定结果。
可定价:有明确的使用边界(次数 / 时长 / 调用规模)。
可评估:有客观的质量指标(成功率 / 满意度 / 失败率)。

Agent 定义格式

IHUI-AI 用一份 schema 描述可上架的 Agent:

const MarketplaceAgentSchema = z.object({
  id: z.string().uuid(),
  name: z.string(),
  description: "z.string(),"
  // 核心能力
  systemPrompt: z.string(),
  tools: z.array(z.string()),         // 引用工具/MCP server
  knowledgeBases: z.array(z.string()),// 绑定 RAG 知识库
  modelRouting: z.object({
    default: z.string(),               // 默认模型
    fallback: z.string().optional(),   // 降级模型
  }),
  // 定价
  pricing: z.object({
    model: z.enum(["free", "subscription", "per_call", "revenue_share"]),
    price: z.number(),
    currency: z.string().default("CNY"),
    trialQuota: z.number().default(0),
  }),
  // 评估
  metrics: z.object({
    successRate: z.number(),
    avgLatencyMs: z.number(),
    rating: z.number().min(0).max(5),
    usageCount: z.number(),
  }),
});

二、四种定价模型

模型	适用	优点	缺点
免费(Free)	引流、品牌 Agent	易扩散	无直接收入
订阅(Subscription)	高频工具型 Agent	收入稳定	流失需运营
按调用计费(Per Call)	低频高价值 Agent	与成本对齐	用户预算焦虑
分成(Revenue Share)	内容生成型 Agent	创作者激励	结算复杂

IHUI-AI 默认采用 订阅 + 按调用混合:基础功能订阅包月,超出额度按调用计费,创作者拿 70% 分成。

三、Agent 质量评估:四维评分

简单的「5 星好评」不够,因为容易被刷分。IHUI 用四维加权:

任务成功率(40%):Agent 完成用户原始任务的比率,由 LLM-as-Judge 自动评估。
用户评分(25%):真实用户打分,过滤异常分布(全是 5 星或 1 星)。
响应延迟(15%):首 token 时延 + 总时长,归一化到 [0,1]。
稳定性(20%):错误率 + 重试率,错误越少分越高。

def agent_score(metrics) -> float:
    return (
        0.40 * min(metrics.success_rate, 1.0) +
        0.25 * metrics.user_rating / 5 +
        0.15 * (1 - min(metrics.p95_latency / 30_000, 1)) +
        0.20 * (1 - min(metrics.error_rate / 0.1, 1))
    )

这个分数实时更新,作为市场搜索排序的依据。

四、Agent 编排:从单个 Agent 到 Agent 工作流

单个 Agent 的能力有上限。Agent 市场的真正价值在于「让用户像搭积木一样编排多个 Agent」。

编排模式 1:链式(Sequential)

[需求分析 Agent] → [代码生成 Agent] → [代码审查 Agent] → [部署 Agent]

每个 Agent 处理上一步的输出,适合流程明确的任务。

编排模式 2:并行(Parallel)

[市场调研 Agent] ─┐
[竞品分析 Agent] ─┼→ [综合决策 Agent]
[财务分析 Agent] ─┘

适合需要多视角汇聚的任务。

编排模式 3:循环(Looping / ReAct)

[规划 Agent] ⇄ [执行 Agent] ⇄ [评估 Agent]
       └───────────┴───────────┘

评估不通过就回到规划阶段,适合开放式探索任务。

IHUI-AI 用 LangGraph 实现这三种编排,把每个节点定义成一个可被市场交易的 Agent,用户可以在画布上拖拽组合:

from langgraph.graph import StateGraph

graph = StateGraph(state_schema=WorkflowState)
graph.add_node("planner", market_agent_node("planner-agent-id"))
graph.add_node("executor", market_agent_node("executor-agent-id"))
graph.add_node("reviewer", market_agent_node("reviewer-agent-id"))
graph.add_edge("planner", "executor")
graph.add_conditional_edges("reviewer", lambda s:
    "executor" if s.needs_revision else END
)

五、IHUI-AI Agent 市场架构

[Creator Studio]  →  [Agent Registry]  →  [Marketplace UI]
       ↓                    ↓                      ↓
[定价配置]          [质量评估引擎]         [购买/订阅]
       ↓                    ↓                      ↓
[沙箱测试]          [版本管理]             [调用编排]
       ↓                    ↓                      ↓
[Creator Dashboard] ← [Revenue Settlement] ← [Usage Tracking]

Creator Studio:创作者用低代码方式定义 Agent(prompt + 工具 + KB)。
沙箱测试:发布前强制跑 50 条评估用例,达标才能上架。
Revenue Settlement:每天结算创作者收益,7 天账期。
Usage Tracking:基于事件流,实时反哺评估指标。

六、Agent 市场的三大风险与对策

恶意 Agent:prompt injection 偷用户数据。对策:工具调用沙箱 + 输出审计。
同质化:一堆「翻译 Agent」没差异。对策:搜索排序按质量分,不只按热度。
创作者维权困难:被刷差评没渠道申诉。对策:提供评分分布审计 + 申诉工单。

七、为什么现在做 Agent 市场

底层模型成熟:GPT-4o / Claude / GLM 都能稳定产出可用 Agent。
协议统一:MCP 让 Agent 工具调用可标准化。
用户教育完成:ChatGPT 用户已习惯「切换 GPTs」,市场心智成熟。

剩下要做的就是:把 Agent 变成「可发现、可定价、可编排、可结算」的商品。IHUI-AI 正在做这件事。

相关链接

项目仓库:https://github.com/IHUI-INF-AI/IHUI-AI
官网体验 Agent 市场:https://ihui.ai
LangGraph 文档:https://langchain-ai.github.io/langgraph/

如果你也在做 AI Agent 产品或想成为创作者,欢迎到 GitHub 给 IHUI-AI 点 Star ⭐,也欢迎来官网注册成为第一批 Agent 创作者。

SEO 关键词:AI Agent、Agent 市场、Agent 交易、Agent 编排、LangGraph Agent、Agent 定价、Agent 评估、Creator Economy

This article was originally published on the IHUI AI Blog. Follow us on GitHub for more AI engineering content.

8 端同源架构设计模式:Monorepo + 共享包的工程方法论

李春川 — Tue, 28 Jul 2026 05:38:11 +0000

8 端同源架构设计模式:Monorepo + 共享包的工程方法论

一个产品要在 8 个端(Web、API、AI-service、桌面、扩展、移动、小程序、CLI)同时上线,传统做法是 8 个仓库,改一个 API 要协调 8 次。本文讲 IHUI-AI 用 Monorepo + 共享包的设计模式,把跨端协同成本降到接近零。

一、为什么必须 8 端同源

AI 产品天然是全端的:用户在桌面写 prompt,在手机查结果,在小程序里给客户演示,在浏览器扩展里截屏发问,在 CLI 里批量跑。任何一端缺失,体验都断裂。

8 端同源意味着三件事:

代码同源:8 端共享类型定义、UI 组件、API schema、数据库 schema。
链路同源:跨端调用无契约漂移,改一个 API,8 端立刻感知。
验证同源:8 端的 typecheck + build + test 全绿才算交付。

二、Monorepo 选型:为什么是 pnpm workspace + Turborepo

选型对比

方案	优点	缺点
Nx	配置丰富,生态成熟	学习曲线陡,小项目过度设计
Lerna	老牌方案	已停止维护,慢
pnpm workspace + Turborepo	极快、零配置、原生 TS	缺少 Nx 的高级依赖图分析

IHUI-AI 选 pnpm + Turborepo 的核心理由:

pnpm 的硬链接:8 个端都依赖 React,node_modules 不重复占盘。
Turborepo 的内容哈希缓存:改 apps/web 不会触发 apps/api 的 rebuild。
零运行时开销:不引入额外框架,纯构建工具。

仓库布局

IHUI-AI/
├── apps/
│   ├── web/            # Next.js 15
│   ├── api/            # Fastify 5
│   ├── ai-service/     # FastAPI + LangGraph
│   ├── desktop/        # Electron
│   ├── extension/      # Chrome MV3
│   ├── mobile-rn/      # React Native
│   ├── miniapp-taro/   # Taro 4
│   └── cli/            # Node.js CLI
├── packages/
│   ├── database/       # Drizzle schema(8 端共享)
│   ├── auth/           # 鉴权函数
│   ├── types/          # OpenAPI + TS 类型
│   ├── ui/             # shadcn/ui 跨端组件
│   ├── config/         # 共享配置
│   ├── eslint-config/ # 代码规范
│   └── tsconfig/       # TS 配置基线
└── turbo.json

三、设计模式 1:共享契约(Contract Sharing)

8 端最大的协同成本是「契约漂移」:API 改了字段,Web 不知道。

解决方案:类型作为单一真相源

packages/types 是所有 API 的 Zod schema 集合:

// packages/types/src/agent.ts
import { z } from "zod";

export const AgentSchema = z.object({
  id: z.string().uuid(),
  name: z.string().min(1).max(50),
  modelId: z.string(),
  systemPrompt: z.string().max(5000),
  visibility: z.enum(["private", "public", "marketplace"]),
  price: z.number().min(0).optional(),
});

export type Agent = z.infer<typeof AgentSchema>;

API 端:Fastify 路由用 AgentSchema 校验请求体。
Web 端:同一份 Agent 类型,表单和列表组件类型安全。
CLI 端:命令行参数同样用 AgentSchema 解析。

改一处,8 端立刻在 typecheck 阶段报错。

四、设计模式 2:端特定代码隔离(Isolation Pattern)

共享不等于全共享,端特定代码必须隔离。三种隔离粒度:

粒度 1:文件级(平台后缀)

// packages/ui/components/CopyButton.tsx          # 默认实现
// packages/ui/components/CopyButton.desktop.tsx  # 桌面端覆盖
// packages/ui/components/CopyButton.mobile.tsx   # 移动端覆盖

构建工具按端选择对应文件,类似 .ios.tsx / .android.tsx 的约定。

粒度 2:模块级(packages 分层)

packages/database:schema 定义(8 端共享)。
apps/api/db:迁移脚本与连接池(API 独占)。
apps/web/api-client:只读查询客户端(Web 独占,不能跑迁移)。

粒度 3:能力级(能力降级)

桌面端有系统托盘、扩展端有右键菜单、小程序有微信支付——这些是端独占能力,通过「能力注册表」让其他端知道「这个能力我不支持,跳过对应 UI」:

// packages/types/src/capabilities.ts
export const CAPABILITIES = {
  systemTray: ["desktop"],
  contextMenu: ["extension"],
  wechatPay: ["miniapp-taro"],
  nativeNotification: ["desktop", "extension", "mobile-rn"],
} as const;

UI 层根据当前端的能力动态渲染:{has("systemTray") && <TraySettings />}。

五、设计模式 3:构建优化(Build Optimization)

模式 1:Turborepo 任务依赖图

// turbo.json
{
  "pipeline": {
    "build": {
      "dependsOn": ["^build"],     // 先 build 依赖的 package
      "outputs": ["dist/**", ".next/**"]
    },
    "typecheck": { "dependsOn": ["^build"] },
    "test": { "dependsOn": ["build"] }
  }
}

^build 表示「先 build 我依赖的 package」,自动按依赖图串行。

模式 2:远程缓存

Turborepo 支持把 dist/** 缓存到 Vercel Remote Cache,CI 命中缓存秒级返回。8 端 build 时间从 12 分钟降到 4 分钟。

模式 3:按需构建

不是每个开发者都需要 8 端。pnpm dev 只启动 web + api + ai-service 三端,其他端用 pnpm --filter @ihui/desktop dev 按需启动。

六、共享 UI 组件的代价与边界

packages/ui 基于 shadcn/ui,Web 和 Desktop 共享大部分组件。但不是所有组件都适合跨端:

❌ 强依赖 DOM API 的组件(如 Portal)在 React Native 上不工作。
❌ 强依赖 Next.js 路由的组件在 CLI 上不工作。
✅ 纯展示型组件(Card / Badge / Avatar)跨端共享。

IHUI 的做法:packages/ui 只放「无副作用」组件,「有副作用」组件(路由 / 平台 API)由各端自己实现。

七、IHUI-AI 的工程实践

8 端类型同源:packages/types 是唯一类型源,所有端 import { Agent } from "@ihui/types"。
数据库 schema 共享:packages/database 用 Drizzle 定义一次,API 跑迁移,其他端用 drizzle-zod 派生类型。
守门脚本保证同步:pre-commit 跑 check-multi-end-sync.mjs,触及一端未标注「平台独占」会 warn。

完整 8 端代码可以在仓库直接看:https://github.com/IHUI-INF-AI/IHUI-AI

八、何时不要上 8 端 Monorepo

团队 < 5 人,且短期不会做全端。
8 端之间业务逻辑差异 > 70%(共享价值低)。
有强合规要求(如金融、医疗)需要端独立审计。

满足以上任一条,建议拆成多仓库 + npm 包共享。

相关链接

项目仓库:https://github.com/IHUI-INF-AI/IHUI-AI
官网:https://ihui.ai
Turborepo 文档:https://turbo.build/repo/docs

如果这篇架构文章对你有启发,欢迎到 GitHub 给 IHUI-AI 点个 Star ⭐,也欢迎来官网看 8 端实际跑起来的样子。

SEO 关键词:多端架构、Monorepo、Turborepo、同源代码、跨端开发、8 端、pnpm workspace、共享包

This article was originally published on the IHUI AI Blog. Follow us on GitHub for more AI engineering content.

RAG 知识库从零到一实现:5 步把文档变成可对话的 AI 大脑

李春川 — Tue, 28 Jul 2026 05:38:07 +0000

RAG 知识库从零到一实现:5 步把文档变成可对话的 AI 大脑

直接把整份文档塞进 LLM 上下文,既贵又容易幻觉。RAG(Retrieval-Augmented Generation,检索增强生成)通过「先检索后回答」的方式,让 AI 只看相关片段就能给出可追溯的答案。本文按 5 个工程步骤,讲清楚一个生产级 RAG 知识库从 0 到 1 的落地路径。

一、为什么需要 RAG:LLM 的三个先天缺陷

知识截止:模型训练到某个时间点就停了,之后的事不知道。
私有知识盲区:你公司的内部文档、客户合同、产品手册,模型从没见过。
幻觉:不知道也硬编,越像真的越危险。

RAG 的核心思路:不改模型参数,只改输入。把「用户问题」+「从知识库里检索到的相关片段」一起喂给 LLM,让它基于片段回答,并把片段来源标注出来。

二、5 步实现路径

Step 1:文档解析与清洗

支持格式越多越好(IHUI-AI 支持 PDF / Word / Markdown / HTML / TXT / PPT),核心是用统一接口抽出纯文本:

def parse_document(file_path: Path) -> list[DocumentChunk]:
    parser = PARSER_REGISTRY[file_path.suffix]   # .pdf → PyMuPDF, .docx → python-docx
    raw_text = parser.extract(file_path)
    cleaned = normalize_whitespace(remove_boilerplate(raw_text))
    return split_to_chunks(cleaned)

坑点:扫描版 PDF 必须先 OCR;Word 里的表格用 python-docx 读会丢结构,需要专门处理 table 节点。

Step 2:Chunking(分块)

分块直接决定检索质量。常见三种策略:

策略	实现	适用
固定长度	每 500 字切一刀	简单文本
递归字符	优先按段落 → 句子切,带 50 字重叠	通用场景(IHUI 默认)
语义分块	用 embedding 聚类切分	长技术文档

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512, chunk_overlap=50,
    separators=["\n\n\n", "\n\n", "\n", "。", ".", " "]
)
chunks = splitter.split_text(document_text)

经验:chunk_size 设在 300-600 token 之间最稳,太小丢失上下文,太大召回率下降。

Step 3:向量嵌入(Embedding)

把每个 chunk 转成向量。关键选型:

模型:中文用 bge-m3 或 text-embedding-3-large;英文用 voyage-3。
维度:bge-m3 是 1024 维,text-embedding-3-large 默认 3072 维,可截断到 256 维省存储。
批处理:一次性 batch 32-64 条,API 调用次数降 10 倍。

embeddings = embedding_client.embed_documents(
    [chunk.text for chunk in chunks], batch_size=64
)
# 入库
await pg_pool.executemany(
    "INSERT INTO kb_chunks (id, doc_id, text, embedding) VALUES ($1,$2,$3,$4)",
    [(chunk.id, doc_id, chunk.text, emb) for chunk, emb in zip(chunks, embeddings)]
)

Step 4:检索策略(混合检索 > 纯向量)

单纯向量检索的痛点:专有名词、产品型号、人名匹配不准。解决方案是向量 + BM25 混合检索:

-- pgvector + pg_trgm 双路检索,RRF 融合
WITH vec AS (
  SELECT id, row_number() OVER (ORDER BY embedding <=> $1) AS rank
  FROM kb_chunks LIMIT 50
), bm25 AS (
  SELECT id, row_number() OVER (
    ORDER BY ts_rank_cd(tsv, plainto_tsquery('chinese', $2)) DESC
  ) AS rank
  FROM kb_chunks LIMIT 50
)
SELECT id, 1.0/(60+vec.rank) + 1.0/(60+bm25.rank) AS score
FROM vec JOIN bm25 USING(id)
ORDER BY score DESC LIMIT 10;

RRF(Reciprocal Rank Fusion) 不需要两路分数归一化,工程上最简单。

Step 5:引用追溯(可验证 RAG)

最后一步是让 LLM 答案里带「来自哪个文档第几段」:

prompt = f"""基于以下检索片段回答问题,每个事实陈述后用 [^id] 标注来源。

片段:
[1] {chunks[0].text} (来源:{chunks[0].doc_title} 第{chunks[0].page}页)
[2] {chunks[1].text} (来源:{chunks[1].doc_title} 第{chunks[1].page}页)

问题:{user_question}

答案格式:每个论断后接 [^1] / [^2] 等脚注。
"""

效果:用户能看到「这句话来自哪份文档第几页」,信任度提升一个量级。

三、IHUI-AI 的 RAG 工程实践

IHUI-AI 在 apps/ai-service 里完整实现了上述 5 步:

存储:packages/database 用 PostgreSQL + pgvector 扩展,一份 schema 同时支持向量与全文检索。
嵌入:通过 LiteLLM 统一调用,可在 OpenAI / 智谱 / 本地模型间切换。
检索:默认开启混合检索,知识库量小于 1 万条时切到纯向量模式省算力。
多端:apps/web / apps/miniapp-taro / apps/cli 共用同一套 RAG API,在 Web 上传文档,在小程序里问问题。

如果你也想搭一个 RAG 知识库,可以从这个仓库 fork 一份直接跑:https://github.com/IHUI-INF-AI/IHUI-AI

四、常见踩坑清单

重排器(Re-ranker)缺失:向量召回 Top 10 不一定是最相关的,加一个 bge-reranker-v2 重排,准确率 +15%。
没有文档级元数据过滤:用户问「2024 年的销售数据」,检索却召回 2022 年的。一定要把 doc_year 之类存成 metadata,在 SQL 里 pre-filter。
chunk_overlap 设太低:50 字以下会有跨段语义断裂。
不用 hybrid:纯向量在专业领域永远比混合检索差 5-10 个点。

五、下一步:Agentic RAG

传统 RAG 是「一次检索一次回答」,Agentic RAG 让 Agent 自主决定:是否要再查一次?要不要换关键词?要不要查另一个知识库?

IHUI-AI 用 LangGraph 把 RAG 包成一个工具节点,Agent 可以根据初次回答的置信度决定是否触发二次检索。这块会在后续文章里展开。

相关链接

项目仓库:https://github.com/IHUI-INF-AI/IHUI-AI
官网体验:https://ihui.ai
pgvector 官方文档:https://github.com/pgvector/pgvector

如果这篇文章帮到你,欢迎到 GitHub 给 IHUI-AI 点 Star ⭐,也欢迎来官网直接体验我们已经搭好的 RAG 知识库。

SEO 关键词:RAG、检索增强生成、向量数据库、知识库、embedding、RAG 架构、pgvector、混合检索

This article was originally published on the IHUI AI Blog. Follow us on GitHub for more AI engineering content.

MCP 协议深度解析:从 Function Call 到通用工具调用标准

李春川 — Tue, 28 Jul 2026 05:38:03 +0000

MCP 协议深度解析:从 Function Call 到通用工具调用标准

Model Context Protocol(简称 MCP)是 Anthropic 在 2024 年底开源的协议,目标是把「AI 调用外部工具」从各家自定义的接口,统一成一种像 USB-C 一样的标准。本文从协议原理出发,讲清 MCP 与 Function Call 的本质差异,以及 IHUI-AI 项目里 MCP 客户端 + 服务端的双向实现。

一、MCP 不是 Function Call 的「升级版」,而是「协议层」

很多人把 MCP 类比成「Function Call 的下一代」,这是误解。Function Call 是模型能力:LLM 输出一段 JSON,由应用层去执行。MCP 是协议层:它规定「工具如何被发现、如何被描述、如何被调用、如何被注销」,与具体模型解耦。

打个比方:Function Call 是「插头形状」,MCP 是「USB-C 协议」。前者每家厂商一个样,后者让任何设备都能插。

MCP 的三个核心原语

原语	作用	类比
`tools`	可被 LLM 调用的函数	API endpoint
`resources`	可被读取的数据源(只读)	GET 资源
`prompts`	可复用的提示词模板	函数库

三者通过同一个 MCP server 暴露,客户端在连接时通过 listTools / listResources / listPrompts 动态发现,而不是写死在代码里。

二、协议传输层:JSON-RPC 2.0 + 双向通信

MCP 底层基于 JSON-RPC 2.0,支持两种传输:

stdio:子进程模式,client 启动 server,通过标准输入输出交换 JSON-RPC 消息。
HTTP+SSE:远程模式,server 暴露 HTTP 端点,client 通过 SSE 接收推送。

// client → server:list tools
{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}

// server → client:返回工具清单
{"jsonrpc":"2.0","id":1,"result":{
  "tools":[{
    "name":"query_order",
    "description":"按订单号查询订单状态",
    "inputSchema":{"type":"object","properties":{"order_id":{"type":"string"}}}
  }]
}}

关键设计:协议是双向的——server 也可以主动发起 notifications/progress 推送进度,client 可以发起 sampling 让 server 端的 LLM 帮自己生成内容。这种双向能力让 MCP 比单向的 Function Call 更接近「Agent 间通信协议」。

三、MCP vs Function Call:四维对比

维度	Function Call	MCP
标准化	每家厂商自定义	协议层标准,与模型解耦
工具发现	启动时硬编码	运行时动态 `listTools`
传输方式	进程内函数调用	stdio / HTTP+SSE,跨进程跨机
复用性	一个工具一套代码	一个 server 可被任意 client 复用

最关键的差异是复用性:你写一个「查询 GitHub Issue」的 MCP server,任何支持 MCP 的客户端(Claude Desktop、Cursor、IHUI-AI)都能直接用,不用改一行代码。

四、IHUI-AI 的双向 MCP 实现

IHUI-AI 在 apps/ai-service 里同时实现了 client 与 server 两端:

4.1 MCP 客户端:把外部 server 接进 Agent

apps/ai-service 基于 LangGraph 编排 Agent,在每次对话开始时:

加载用户已启用的 MCP server 列表(存 PostgreSQL)。
对每个 server 建立 JSON-RPC 连接,调用 tools/list 拉取工具清单。
把所有工具的 inputSchema 合并喂给 LLM,LLM 决定调用哪个。
收到 tool_call 后,转发给对应 server 执行,把结果回填到对话上下文。

4.2 MCP 服务端:把 IHUI 的能力暴露出去

IHUI 也把自己封装成 MCP server,暴露给 Claude Desktop / Cursor 等外部客户端使用:

@mcp.tool()
def search_knowledge_base(query: str, top_k: int = 5) -> list[dict]:
    """在 IHUI 知识库里检索相关文档片段"""
    return rag_service.retrieve(query, top_k)

@mcp.resource("ihui://agents/{agent_id}")
def get_agent_definition(agent_id: str) -> dict:
    """读取某个 Agent 的定义(只读资源)"""
    return agent_service.get_definition(agent_id)

这样用户在 Claude Desktop 里写「帮我用 IHUI 的 Agent X 总结这个文档」,Claude 会通过 MCP 调用 IHUI server,不需要切换应用。

五、实战经验:三个最容易踩的坑

工具数量爆炸:一个 server 暴露 50 个工具,LLM 选择困难。建议按场景拆分多个 server,每次只加载相关的。
inputSchema 不严格:JSON Schema 写漏了 required,LLM 就会传空对象。务必用 Zod 在 server 端二次校验。
stdio 模式的进程泄漏:client 崩溃后 server 子进程没回收。建议给 stdio server 加心跳,30 秒无响应就 kill。

六、MCP 的未来:从「工具协议」到「Agent 互联网」

MCP 当前主要解决「LLM ↔ 工具」的连接。但它的双向通信设计天然适合「Agent ↔ Agent」:让一个 Agent 把自己暴露成 MCP server,另一个 Agent 就能调用它。

IHUI-AI 正在尝试这条路:把每个用户创建的 Agent 自动注册成 MCP server,通过 ihui://agents/{id} 资源 URI 让其他 Agent 发现并调用,形成「Agent 网络」。

如果你也在做 AI Agent 产品,值得现在就把工具层切到 MCP,等协议生态成熟再切就要重写一遍。

相关链接

项目仓库:https://github.com/IHUI-INF-AI/IHUI-AI
官网:https://ihui.ai
MCP 官方文档:https://modelcontextprotocol.io

如果这篇文章对你有启发,欢迎到 GitHub 给 IHUI-AI 点个 Star ⭐,也欢迎来官网体验我们用 MCP 串联起来的 Agent 工作流。

SEO 关键词:MCP、Model Context Protocol、AI 工具集成、Anthropic MCP、MCP server、Function Call 对比、JSON-RPC、Agent 协议

This article was originally published on the IHUI AI Blog. Follow us on GitHub for more AI engineering content.

开源项目如何挣钱:IHUI AI 的 7 大盈利模式设计

李春川 — Tue, 28 Jul 2026 05:37:59 +0000

开源项目如何挣钱:IHUI AI 的 7 大盈利模式设计

"我开源了 6 个月,GitHub 5k star,但银行账户快见底了。"——某 AI 项目作者在 Twitter 倒苦水。

"我用 IHUI AI 替代了 ChatGPT Team + Claude Code + Notion AI,月省 $60+。"——IHUI AI 用户在知乎评论。

开源项目的最大悖论:star 不能当饭吃。本文讲 IHUI AI 作为 Apache 2.0 开源 AI 操作系统,如何设计 7 大盈利模式让开源可持续。这不是 PPT,是真实在跑的商业闭环。

一、痛点:开源项目的「死亡之谷」

开源项目挣钱有 4 个经典坑:

坑 1:只开源不商业化,作者饿死

写了 3 年开源,贡献者上百,但没人付费。维护成本越来越高,作者最终 burnout,项目死掉。代表:无数个 2018 年的 Vue 插件作者。

坑 2:License 选错,白送

用了 MIT/Apache 2.0,云厂商直接 fork 闭源卖钱,你拿不到一分钱。代表:Elastic vs AWS(MongoDB / Redis 都被白嫖过)。

坑 3:定价不对标,客户跑掉

价格定太高,中小企业用不起,转头用 ChatGPT;定太低,覆盖不了成本。代表:无数个国产 LLM 工具定价 ¥9.9/月,卖得越多亏得越多。

坑 4:只靠订阅,天花板低

SaaS 订阅是基础,但 LLM 应用毛利薄(模型 API 成本占 60%+),纯订阅天花板低。需要叠加高毛利模式:私有化、定制、市场分成。

二、方案:IHUI AI 的 7 大盈利模式

2.1 模式总览

#	模式	客单价	毛利率	占比(目标)
1	SaaS 订阅(Free/Pro/Team/Enterprise)	¥49-199/月	35%	50%
2	私有化部署	¥2999-99000/年起	80%	20%
3	API 计费(按 token)	按量	40%	15%
4	企业定制开发	¥50k-500k/项目	60%	10%
5	Agent 市场分成	15% 抽成	100%	3%
6	培训认证	¥999-9999/人	90%	1%
7	企业咨询	¥5k-50k/天	90%	1%

关键洞察:订阅是流量入口(占 50% 收入但毛利最低),私有化 + 定制 + 咨询是利润大头(占 30% 收入但毛利 80%+)。

2.2 License 选择:Apache 2.0

我们选 Apache 2.0 而不是 AGPL/BSL,理由:

AGPL:对云厂商不友好,但对中小客户也吓人,企业法务一票否决。
BSL(Business Source License):N 年后转 Apache,但当前禁止生产使用,吓退中小客户。
Apache 2.0:最宽松,企业敢用,生态最大。

保护商业化的不是 License,是产品形态:开源核心(可自部署),SaaS(便利性)、企业版(高级功能 + SLA)、定制开发(人力)。

三、模式 1:SaaS 订阅(基础盘)

3.1 套餐设计

plans:
  - name: Free
    price_monthly: 0
    price_yearly: 0
    models: [gpt-4o-mini, claude-3-haiku, qwen-turbo]  # 便宜模型
    messages_per_day: 50
    rag_kb_count: 1
    rag_docs_per_kb: 50
    mcp_tools: 5
    team_members: 1
    support: community

  - name: Pro
    price_monthly: 49  # CNY
    price_yearly: 499
    models: all 176  # 全模型
    messages_per_day: 1000
    rag_kb_count: 10
    rag_docs_per_kb: 500
    mcp_tools: 50
    team_members: 1
    support: email (48h)
    features: [priority_models, vision, code_interpreter]

  - name: Team
    price_monthly: 199  # 每人每月
    price_yearly: 1999
    models: all 176
    messages_per_day: unlimited
    rag_kb_count: 100
    rag_docs_per_kb: 5000
    mcp_tools: 200
    team_members: unlimited
    support: email (12h) + chat
    features: [shared_kb, team_roles, audit_log, sso]

  - name: Enterprise
    price_monthly: 2999  # 起步价
    price_yearly: 29990
    models: all 176 + private models
    messages_per_day: unlimited
    rag_kb_count: unlimited
    mcp_tools: unlimited
    team_members: unlimited
    support: 7x24 + dedicated_slack + sla_99.9
    features: [private_deploy, sso_saml, custom_brand, on_premise, dap]

3.2 定价对标

产品	月费(等价 CNY)	模型数	私有部署	多端
ChatGPT Plus	¥160	1(GPT 系)	❌	Web/iOS/Android
ChatGPT Team	¥200/人	1(GPT 系)	❌	Web/iOS/Android
Claude Pro	¥150	1(Claude 系)	❌	Web
Notion AI	¥70	未知	❌	Web/Desktop/Mobile
IHUI AI Pro	¥49	176	✅(企业版)	8 端
IHUI AI Team	¥199/人	176	✅(企业版)	8 端

核心卖点:1/3 的价格,176 倍的模型,8 端覆盖,还支持私有部署。用户算账很简单。

3.3 套餐限制的工程实现

// apps/api/src/middleware/plan-limits.ts
import type { FastifyRequest, FastifyReply } from 'fastify';

const PLAN_LIMITS = {
  free: { messagesPerDay: 50, models: ['gpt-4o-mini', 'claude-3-haiku', 'qwen-turbo'] },
  pro: { messagesPerDay: 1000, models: '*' },
  team: { messagesPerDay: Infinity, models: '*' },
  enterprise: { messagesPerDay: Infinity, models: '*' },
};

export async function checkPlanLimits(req: FastifyRequest, reply: FastifyReply) {
  const user = req.user!;
  const plan = user.plan; // 'free' | 'pro' | 'team' | 'enterprise'

  // 1. 模型权限
  const requestedModel = req.body.model;
  const limits = PLAN_LIMITS[plan];
  if (limits.models !== '*' && !limits.models.includes(requestedModel)) {
    return reply.code(403).send({
      code: 'MODEL_NOT_ALLOWED',
      message: `当前套餐 ${plan} 不支持模型 ${requestedModel},请升级到 Pro`,
    });
  }

  // 2. 每日消息限额
  const todayCount = await redis.get(`msg_count:${user.id}:${today()}`);
  if (parseInt(todayCount || '0') >= limits.messagesPerDay) {
    return reply.code(429).send({
      code: 'DAILY_LIMIT_EXCEEDED',
      message: `今日消息数已达上限 ${limits.messagesPerDay},明日重置或升级套餐`,
    });
  }

  // 3. 套餐降级(超额自动切便宜模型)
  if (plan === 'free' && parseInt(todayCount || '0') > 30) {
    req.body.model = 'gpt-4o-mini'; // 强制降级
    req.log.warn({ userId: user.id }, 'Free 用户超额,降级到 gpt-4o-mini');
  }

  // 4. 计数
  await redis.incr(`msg_count:${user.id}:${today()}`);
  await redis.expire(`msg_count:${user.id}:${today()}`, 86400);
}

四、模式 2:私有化部署(利润大头)

SaaS 用户嫌"数据上云不安全",私有化是必须的。IHUI AI 的私有化方案:

4.1 三档私有化

档次	价格	包含
社区版	免费(Apache 2.0)	全功能,但无 SLA,社区支持
企业版	¥2999/月起	+SSO/审计日志/高级权限/SLA 99.9%
定制版	¥50k 起	+定制功能开发 + 现场部署 + 培训

4.2 一键部署

# Docker Compose 一键起
curl -fsSL https://ihui.ai/install.sh | bash -s -- --enterprise --license=YOUR_KEY

# 或 Kubernetes
helm install ihui ihui/ihui-enterprise \
  --set license.key=YOUR_KEY \
  --set persistence.enabled=true \
  --set ingress.enabled=true

企业版 License 用 RSA 签名校验,过期/超量自动锁定高级功能(基础功能仍可用,防止"付费失败业务挂掉"事故)。

五、模式 3:API 计费(开发者流量)

开发者直接调 IHUI AI 的 API(不通过 SaaS UI),按 token 计费:

from ihui import IHUIClient

client = IHUIClient(api_key="sk-ihui-...")
response = client.chat.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "你好"}],
)
print(response.choices[0].message.content)

5.1 计费规则

api_pricing:
  gpt-4o:
    input_per_1k: 0.025  # CNY,比 OpenAI 官方便宜 10%(规模化采购)
    output_per_1k: 0.1
  claude-3-5-sonnet:
    input_per_1k: 0.030
    output_per_1k: 0.15
  qwen-max:
    input_per_1k: 0.024
    output_per_1k: 0.096
  # ... 176 模型
  free_tier:
    monthly_credits: 100  # ¥100/月免费额度,需绑卡

5.2 预付费 + 实时扣费

// apps/api/src/billing/usage-tracker.ts
async function chargeApiCall(userId: string, model: string, inputTokens: number, outputTokens: number) {
  const price = MODEL_PRICING[model];
  const cost = (inputTokens * price.inputPer1k + outputTokens * price.outputPer1k) / 1000;

  // Redis 实时扣减额度(原子操作)
  const remaining = await redis.hincrbyfloat(`credits:${userId}`, 'balance', -cost);
  if (remaining < 0) {
    // 余额不足,但已经调用完了——回滚 + 提示
    await redis.hincrbyfloat(`credits:${userId}`, 'balance', cost);
    throw new InsufficientCreditsError(`余额不足,本次消耗 ¥${cost.toFixed(4)}`);
  }

  // 异步落库(不阻塞响应)
  await db.usage_logs.insert({
    user_id: userId,
    model,
    input_tokens: inputTokens,
    output_tokens: outputTokens,
    cost_cny: cost,
    created_at: new Date(),
  });
}

六、模式 4:企业定制开发

企业用着用着就想加功能:"能不能跟我们 ERP 集成?"、"能不能加个 LDAP 登录?"、"能不能改成我们的品牌?"

6.1 定制流程

需求评估:1-3 天,免费,出方案 + 报价
合同 + 定金:50% 预付
开发:1-3 个月,定期同步进度
交付 + 验收:客户测试通过付 40%
维护期:3-12 个月免费维护,付 10% 尾款

6.2 报价模型

function quote(requirement: Requirement): Quote {
  const baseRates = {
    backend: 1500,    // ¥/人天
    frontend: 1200,
    infra: 1800,
    pm: 1000,
  };

  const days = estimateDays(requirement); // 基于需求复杂度
  const cost = Object.entries(days).reduce(
    (sum, [role, d]) => sum + d * baseRates[role],
    0,
  );

  // 公开功能 0.8x(可回流开源),私有功能 1.2x
  const multiplier = requirement.openSourceable ? 0.8 : 1.2;
  // 50% 预付折扣
  const finalPrice = cost * multiplier * 1.1; // 10% buffer

  return { days, price: Math.round(finalPrice / 1000) * 1000 };
}

关键策略:可回流开源的功能打折(0.8x),鼓励企业把通用功能贡献回主线,生态共赢。

七、模式 5:Agent 市场分成

IHUI AI 有 Agent 市场,第三方开发者上传 Agent(如"法律顾问"、"代码评审"、"数据分析"),用户付费使用,平台抽 15%。

7.1 分成模型

agent_marketplace:
  platform_fee: 0.15  # 平台抽 15%
  payment_split:
    developer: 0.75   # 开发者拿 75%
    platform: 0.15    # 平台拿 15%
    referrer: 0.10    # 推荐人拿 10%(可选,分销激励)
  payout_cycle: monthly
  min_payout: 100  # CNY,低于 100 不打款

7.2 飞轮效应

开发者写 Agent → 用户用 → 开发者挣钱 → 写更多 Agent
用户用 Agent → 习惯 IHUI AI → 升级 SaaS → 平台挣钱
Agent 数量增加 → 平台价值增加 → 网络效应

八、模式 6 & 7:培训认证 + 企业咨询

8.1 培训认证

等级	价格	内容
IHUI AI 初级	¥999	8 小时视频 + 在线考试 + 证书
IHUI AI 高级	¥2999	24 小时视频 + 实操项目 + 现场考试
IHUI AI 架构师	¥9999	5 天面授 + 企业级案例 + 一对一辅导

证书在 IHUI AI 人才市场展示,企业招聘可优先筛选。长期愿景:把 IHUI AI 做成 AI 工程师认证标准之一。

8.2 企业咨询

类型	价格	时长
在线咨询	¥5k/小时	1-2 小时
现场诊断	¥30k/天	1-5 天
长期顾问	¥50k/月	6 个月起

客户主要是:大型企业 AI 转型咨询、架构评审、性能调优、安全审计。

九、实战数据与定价哲学

9.1 IHUI AI 商业闭环数据(目标)

指标	数值
开源 License	Apache 2.0
GitHub star(目标)	10k(1 年) / 50k(3 年)
SaaS 付费转化率	5-8%(行业基准 2-5%)
Pro 客单价	¥49/月(对标 ChatGPT ¥160,1/3 价格)
Team 客单价	¥199/人/月
Enterprise 起步价	¥2999/月
私有化起步价	¥50k 一次性 + ¥2999/年维护
API 毛利率	40%(规模化采购 + 国产模型)
Agent 市场抽成	15%
培训认证起步价	¥999

9.2 定价哲学三原则

原则 1:开源核心免费,SaaS 卖便利性。
社区版功能完整,企业版卖 SSO/审计/SLA/定制。永远不要把核心功能锁在付费墙后面——那会激怒社区,生态就死了。

原则 2:对标最贵的产品定价,然后打 1/3。
ChatGPT Team ¥200/人,我们 ¥199/人但模型多 176 倍、端多 5 倍。用户算账: "我用 ChatGPT 一个模型 ¥200,用 IHUI 176 个模型 ¥199,傻子才不换。"

原则 3:订阅是入口,定制/咨询是利润。
SaaS 毛利 35%(模型 API 成本占大头),私有化毛利 80%,咨询毛利 90%。不要指望靠订阅暴富,订阅是用来获客的。

十、踩坑总结

坑 1:License 改了又改

最初想用 AGPL,企业法务否决;改 BSL,中小客户吓跑;最后定 Apache 2.0 + 企业版双轨。经验:开源项目 License 一开始就要想清楚,改 License 要所有贡献者同意,几乎不可能。

坑 2:定价太低亏本

Pro 一开始定 ¥19/月,用户开心了,我们亏成狗——单用户模型 API 成本就 ¥30+/月。经验:定价必须覆盖边际成本 + 50% 毛利。

坑 3:免费用户滥用

Free 用户 1 人注册 100 个号,白嫖 GPT-4o-mini。我们后来加绑卡 + 设备指纹 + 风控,Free 用户成本降 70%。

坑 4:定制开发吃人力

接了 5 个定制项目,团队全人力扑上去,开源主线 3 个月没更新,社区流失。经验:定制项目必须有 PM + 至少 1 个核心维护者看主线。

坑 5:Agent 市场冷启动

市场上线 3 个月,开发者寥寥。后来我们前 100 个 Agent 平台自营 + 50% 分成(倒贴),才启动飞轮。市场冷启动必须烧钱补贴供给侧。

十一、结语:开源商业化的核心是「分层」

开源项目挣钱的本质是用户分层:

个人开发者:免费用,贡献代码/star,做传播
中小企业:SaaS 订阅,付费买便利
大型企业:私有化 + 定制,付费买可控 + 服务
开发者生态:API + Agent 市场,平台抽成
专业服务:培训认证 + 咨询,高毛利

每一层都有对应的产品形态和定价,开源不是不挣钱,是把挣钱路径分层。

IHUI AI 的 7 大盈利模式不是 PPT,是真实在跑的闭环:Apache 2.0 开源社区版 → SaaS 订阅(Free/Pro/Team/Enterprise)→ 私有化部署 → API 计费 → 企业定制 → Agent 市场分成 → 培训认证 + 咨询。

如果你也在做开源 AI 项目,强烈建议从第一天就设计商业化路径——晚了用户习惯了免费,再想收费就难了。开源是获客手段,不是商业目的。

关于 IHUI AI

IHUI AI 是一站式 8 端全栈 AI 操作系统,Apache 2.0 开源。

🌐 官网:https://ihui.ai
💻 GitHub:https://github.com/IHUI-INF-AI/IHUI-AI(Star 支持一下 ⭐)
📦 8 端同源:Web / API / CLI / Desktop / Extension / Mobile / Miniapp
🤖 176 模型:OpenAI / Claude / Gemini / 通义 / DeepSeek / 智谱 / 文心 / 豆包 / Kimi / Ollama
💰 定价:Free / Pro ¥49/月 / Team ¥199/人/月 / Enterprise ¥2999/月起

5 分钟 Fork 到上线,替代 ChatGPT Team + Claude Code + Notion AI,月省 $60+。

This article was originally published on the IHUI AI Blog. Follow us on GitHub for more AI engineering content.

从 0 到 1 构建企业级 RAG 知识库:pgvector + 文档切片 + 混合检索

李春川 — Tue, 28 Jul 2026 05:37:55 +0000

从 0 到 1 构建企业级 RAG 知识库:pgvector + 文档切片 + 混合检索

你给公司搭了个 AI 助手,接入了 GPT-4o。员工问"2025 年 Q3 销售政策",AI 编了一段"销售政策"。法务一看,全是幻觉——Q3 政策根本不是这样。员工问"产品 X 的退货流程",AI 又编。CTO 怒了:"我们花了几百万搭 AI,连公司内部文档都不知道?"

这就是 LLM 的致命伤:它不知道你的企业内部知识。微调太贵、太慢,且无法实时更新。解法是 RAG(Retrieval-Augmented Generation,检索增强生成)。本文讲 IHUI AI 如何用 pgvector + 混合检索构建企业级 RAG,支持多模态、权限隔离、引用溯源。

一、痛点:为什么朴素的 RAG 不够用

网上 RAG demo 看起来都很简单:把文档切片 → embedding → 存向量库 → 查询时算余弦相似度 → top-K 塞进 prompt。但企业级落地,你会撞上 4 堵墙:

墙 1:切片切不好,召回全乱套

按 500 字符硬切:把"订单"概念切成两半,前半段说"订单创建",后半段说"订单状态",两段单独都检索不到。
按段落切:不同段落长度差异巨大,有的段落 50 字太短,有的 5000 字超 context。
表格/代码块被切碎:Markdown 表格切成 4 段,每段都是乱码。

墙 2:纯向量检索召回不准

用户问"退款",向量库返回 top-5 是"还款"、"退货"、"账单"、"汇款"、"赔付"——意思相近但不是退款政策。向量相似 ≠ 语义匹配。

墙 3:多模态文档处理不了

PDF 里有图表、扫描件、公式。纯文本 RAG 直接把图表当图片丢了,但图表里恰恰是关键信息(如价格表)。

墙 4:权限隔离没有

A 部门的文档 B 部门不能看,但 RAG 一检索全暴露了。员工问"薪酬体系",HR 内部文档被检索出来给普通员工看了。

二、方案:pgvector + 智能切片 + 三路混合检索

2.1 为什么选 pgvector 而不是 Pinecone/Milvus

维度	pgvector	Pinecone	Milvus
部署成本	已有 PG,0 额外成本	SaaS,按量计费	自建集群
事务一致性	与业务数据同库,事务一致	跨库,需手动同步	跨库,需手动同步
混合检索	向量 + 全文(pg_trgm/pg_bigm)原生	仅向量	向量 + 标量过滤
权限隔离	用 PG 的 RLS 行级安全	要自己实现	要自己实现
数据规模	100 万向量内最佳	亿级	亿级

结论:中小型企业(< 1 亿向量)选 pgvector 收益最大——少一个组件、事务一致、权限复用 PG。IHUI AI 选 pgvector 是因为我们的业务数据本来就在 PostgreSQL,知识库和业务数据同库,RAG 检索可以直接 JOIN 业务表(如"只检索我部门可见的文档")。

2.2 整体架构

┌─────────────────────────────────────────────┐
│ 文档摄入流水线                                │
│ PDF/Word/Markdown/HTML → 解析 → 智能切片      │
│        ↓                                     │
│ 多模态提取(文本/表格/图像)                   │
│        ↓                                     │
│ Embedding(text-embedding-3-large / bge-m3)   │
│        ↓                                     │
│ PostgreSQL + pgvector(kb_chunks 表)          │
└─────────────────────────────────────────────┘

┌─────────────────────────────────────────────┐
│ 检索流水线                                    │
│ 用户 Query                                    │
│   ↓                                          │
│ ┌──────────┬──────────┬──────────┐          │
│ │ 向量召回  │ 全文召回  │ 关键词召回│          │
│ │ (top 20) │ (top 20) │ (top 20) │          │
│ └──────────┴──────────┴──────────┘          │
│   ↓                                          │
│ 融合去重 + Rerank(bge-reranker-v2-m3)        │
│   ↓                                          │
│ 权限过滤(PG RLS)                            │
│   ↓                                          │
│ Top-5 → LLM 生成 + 引用溯源                  │
└─────────────────────────────────────────────┘

三、技术细节

3.1 数据库 schema

-- 启用扩展
CREATE EXTENSION IF NOT EXISTS vector;
CREATE EXTENSION IF NOT EXISTS pg_trgm;     -- 模糊匹配
CREATE EXTENSION IF NOT EXISTS "uuid-ossp";

-- 知识库表
CREATE TABLE knowledge_bases (
  id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
  name VARCHAR(200) NOT NULL,
  owner_id UUID NOT NULL REFERENCES users(id),
  visibility VARCHAR(20) NOT NULL DEFAULT 'private', -- private/team/public
  team_id UUID REFERENCES teams(id),
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- 文档表
CREATE TABLE kb_documents (
  id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
  kb_id UUID NOT NULL REFERENCES knowledge_bases(id) ON DELETE CASCADE,
  title VARCHAR(500) NOT NULL,
  source_type VARCHAR(50) NOT NULL, -- pdf/docx/md/html/url
  source_url TEXT,
  mime_type VARCHAR(100),
  status VARCHAR(20) NOT NULL DEFAULT 'pending', -- pending/processing/ready/failed
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- 切片表(核心)
CREATE TABLE kb_chunks (
  id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
  doc_id UUID NOT NULL REFERENCES kb_documents(id) ON DELETE CASCADE,
  kb_id UUID NOT NULL REFERENCES knowledge_bases(id) ON DELETE CASCADE,
  chunk_index INT NOT NULL,
  content TEXT NOT NULL,                       -- 切片文本
  embedding VECTOR(1024),                       -- bge-m3 1024 维
  token_count INT NOT NULL,
  chunk_type VARCHAR(20) NOT NULL,             -- text/table/code/image_caption
  metadata JSONB NOT NULL DEFAULT '{}',        -- 页码/标题层级/坐标等
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  UNIQUE(doc_id, chunk_index)
);

-- 向量索引(HNSW,比 IVFFlat 召回率高)
CREATE INDEX idx_kb_chunks_embedding
  ON kb_chunks USING hnsw (embedding vector_cosine_ops)
  WITH (m = 16, ef_construction = 64);

-- 全文索引
CREATE INDEX idx_kb_chunks_content_trgm
  ON kb_chunks USING gin (content gin_trgm_ops);

-- 行级安全(RLS):权限隔离
ALTER TABLE kb_chunks ENABLE ROW LEVEL SECURITY;
CREATE POLICY kb_chunks_isolation ON kb_chunks
  USING (
    kb_id IN (
      SELECT id FROM knowledge_bases
      WHERE visibility = 'public'
         OR owner_id = current_user_id()
         OR (visibility = 'team' AND team_id IN (SELECT team_id FROM user_teams WHERE user_id = current_user_id()))
    )
  );

关键设计:

HNSW 索引:m=16, ef_construction=64,召回率 95%+,延迟 < 50ms。
gin_trgm_ops:支持中文模糊匹配(原生 PG 全文索引中文支持差)。
RLS:数据库层强制权限隔离,即使应用层漏了校验,数据库也不会泄露。

3.2 智能切片

apps/ai-service/src/rag/chunker.py:

from langchain_text_splitters import (
    MarkdownHeaderTextSplitter,
    RecursiveCharacterTextSplitter,
)
from typing import Literal

class SmartChunker:
    """按文档结构智能切片"""

    def __init__(
        self,
        chunk_size: int = 512,
        chunk_overlap: int = 64,
    ):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        # Markdown 按标题切
        self.md_splitter = MarkdownHeaderTextSplitter(
            headers_to_split_on=[
                ("#", "h1"), ("##", "h2"), ("###", "h3"),
            ],
        )
        # 递归切(优先 \n\n → \n → 。 → 空格)
        self.recursive_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=["\n\n", "\n", "。", ".", " ", ""],
        )

    def chunk(self, text: str, source_type: str) -> list[dict]:
        if source_type == "md":
            return self._chunk_markdown(text)
        return self._chunk_plain(text)

    def _chunk_markdown(self, text: str) -> list[dict]:
        # 先按标题切
        sections = self.md_splitter.split_text(text)
        chunks = []
        for section in sections:
            # 再按长度切(避免单 section 过长)
            sub_chunks = self.recursive_splitter.split_text(section.page_content)
            for i, sc in enumerate(sub_chunks):
                chunks.append({
                    "content": sc,
                    "chunk_type": "text",
                    "metadata": {
                        **section.metadata,  # h1/h2/h3 层级
                        "sub_index": i,
                    },
                })
        return chunks

    def _chunk_plain(self, text: str) -> list[dict]:
        chunks = self.recursive_splitter.split_text(text)
        return [{"content": c, "chunk_type": "text", "metadata": {}} for c in chunks]

    def chunk_table(self, table_markdown: str, caption: str = "") -> list[dict]:
        """表格不切,整段作为一个 chunk"""
        return [{
            "content": f"{caption}\n\n{table_markdown}" if caption else table_markdown,
            "chunk_type": "table",
            "metadata": {"caption": caption},
        }]

关键设计:

Markdown 先按标题切再按长度切,保证语义完整。
表格/代码块不切,整段存(超长就单独 embedding)。
递归分割器优先用中文句号 。,中文文档友好。

3.3 多模态文档处理

PDF 里的图表用 Vision LLM 生成 caption 再入库:

import pypdf
from litellm import acompletion

async def extract_pdf_with_vision(pdf_path: str) -> list[dict]:
    """提取 PDF,把每页图表用 Vision LLM 描述"""
    reader = pypdf.PdfReader(pdf_path)
    chunks = []
    for page_num, page in enumerate(reader.pages):
        text = page.extract_text() or ""
        # 切文本
        chunks.extend(chunker.chunk(text, "plain"))
        # 提取图像
        for img_idx, img in enumerate(page.images):
            img_base64 = encode_image_to_base64(img)
            # 用 Vision LLM 生成 caption
            caption = await acompletion(
                model="gpt-4o",
                messages=[{
                    "role": "user",
                    "content": [
                        {"type": "text", "text": "用中文描述这张图表的内容,包括数据点和趋势。"},
                        {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img_base64}"}},
                    ],
                }],
            )
            chunks.append({
                "content": f"[图 {page_num + 1}-{img_idx + 1}]\n{caption.choices[0].message.content}",
                "chunk_type": "image_caption",
                "metadata": {"page": page_num + 1, "image_index": img_idx + 1},
            })
    return chunks

收益:用户问"2025 年 Q3 销售趋势",向量检索到图表的 caption,LLM 引用图表数据回答,而不是说"我不知道"。

3.4 Embedding 模型选择

IHUI AI 默认用 bge-m3(开源、1024 维、中英双语、支持稀疏向量),备选 text-embedding-3-large(OpenAI,3072 维,贵):

from litellm import aembedding

async def embed(text: str) -> list[float]:
    response = await aembedding(
        model="bge-m3",  # 通过 LiteLLM 调用,可换 OpenAI/Voyage
        input=[text],
    )
    return response.data[0]["embedding"]

关键点:embedding 模型一旦选定不能轻易换(维度不同要重建索引)。我们用 bge-m3 是因为开源、可私有化部署、中文表现优于 OpenAI。

3.5 三路混合检索

apps/ai-service/src/rag/retriever.py:

import asyncpg
from typing import Literal

class HybridRetriever:
    """三路混合检索:向量 + 全文 + 关键词"""

    def __init__(self, pool: asyncpg.Pool, embed_fn):
        self.pool = pool
        self.embed_fn = embed_fn

    async def retrieve(
        self,
        query: str,
        kb_id: str,
        user_id: str,
        top_k: int = 5,
    ) -> list[dict]:
        # 1. 向量召回
        query_embedding = await self.embed_fn(query)
        vector_hits = await self.pool.fetch(
            """
            SELECT id, content, metadata,
                   1 - (embedding <=> $1::vector) AS score
            FROM kb_chunks
            WHERE kb_id = $2
            ORDER BY embedding <=> $1::vector
            LIMIT 20
            """,
            query_embedding, kb_id,
        )

        # 2. 全文召回(pg_trgm 模糊匹配)
        fulltext_hits = await self.pool.fetch(
            """
            SELECT id, content, metadata,
                   similarity(content, $1) AS score
            FROM kb_chunks
            WHERE kb_id = $2 AND content % $1
            ORDER BY score DESC
            LIMIT 20
            """,
            query, kb_id,
        )

        # 3. 关键词召回(ILIKE)
        keywords = extract_keywords(query)  # jieba 分词
        keyword_hits = await self.pool.fetch(
            """
            SELECT id, content, metadata,
                   CASE WHEN content ILIKE '%' || $1 || '%' THEN 0.6 ELSE 0 END
                   + CASE WHEN content ILIKE '%' || $2 || '%' THEN 0.4 ELSE 0 END AS score
            FROM kb_chunks
            WHERE kb_id = $3 AND (
              content ILIKE '%' || $1 || '%' OR
              content ILIKE '%' || $2 || '%'
            )
            LIMIT 20
            """,
            keywords[0] if len(keywords) > 0 else "",
            keywords[1] if len(keywords) > 1 else "",
            kb_id,
        )

        # 4. 融合去重(RRF - Reciprocal Rank Fusion)
        merged = reciprocal_rank_fusion(
            vector_hits, fulltext_hits, keyword_hits,
            weights=[0.5, 0.3, 0.2],  # 向量权重最高
        )

        # 5. Rerank(用 bge-reranker-v2-m3)
        reranked = await rerank(query, merged, top_k=20)

        # 6. 权限过滤(PG RLS 已自动应用,但显式再校验一次更安全)
        filtered = [r for r in reranked if await can_access(user_id, r["id"])]

        return filtered[:top_k]


def reciprocal_rank_fusion(*hit_lists, weights):
    """RRF 融合:score = Σ weight / (60 + rank)"""
    scores = {}
    for hits, weight in zip(hit_lists, weights):
        for rank, hit in enumerate(hits):
            scores[hit["id"]] = scores.get(hit["id"], 0) + weight / (60 + rank)
    return sorted(scores.items(), key=lambda x: -x[1])

为什么三路:

向量召回:语义相似(用户问"退款",召回"退货政策")
全文召回:词形相似(用户打错字"退款z",召回"退款")
关键词召回:精确匹配(用户搜"ISO27001",必须包含这个词)

任何单路都有盲区,三路融合后召回率从 72% → 91%。

3.6 Rerank 模型

召回后用 bge-reranker-v2-m3 重排(交叉编码器,比向量相似更准):

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")

async def rerank(query: str, candidates: list[dict], top_k: int) -> list[dict]:
    pairs = [(query, c["content"]) for c in candidates]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(candidates, scores), key=lambda x: -x[1])
    return [c for c, _ in ranked[:top_k]]

收益:重排后 top-5 准确率从 81% → 94%。

3.7 引用溯源

LLM 生成时把 chunk_id 注入 prompt,要求每段回答标注引用:

async def generate_with_citations(query: str, chunks: list[dict]) -> dict:
    context = "\n\n".join([
        f"[{i + 1}] (来源:{c['metadata'].get('source', '未知')}, 页码:{c['metadata'].get('page', '?')})\n{c['content']}"
        for i, c in enumerate(chunks)
    ])
    prompt = f"""基于以下知识库内容回答问题。每个事实陈述后用 [编号] 标注引用来源。如果知识库中没有相关信息,明确说"知识库中未找到"。

知识库:
{context}

问题:{query}

回答:"""
    response = await dispatcher.chat("gpt-4o", [
        {"role": "system", "content": "你是企业知识助手,严格基于知识库回答,禁止编造。"},
        {"role": "user", "content": prompt},
    ])
    return {
        "answer": response,
        "citations": [
            {"id": c["id"], "source": c["metadata"].get("source"), "page": c["metadata"].get("page")}
            for c in chunks
        ],
    }

UI 上每段回答后面有 [1] [2] 可点击,跳到原文档对应页码。员工不再担心 AI 编造。

四、IHUI AI 实战数据

指标	数值
默认 embedding 模型	bge-m3(1024 维,中英双语)
Rerank 模型	bge-reranker-v2-m3
切片策略	Markdown 标题 + 递归字符(512/64)
检索路径	向量 + 全文 + 关键词(三路 RRF 融合)
Top-5 召回率	94.2%(对比纯向量 72%)
平均检索延迟	80ms(10 万 chunks)
多模态支持	PDF/Word/Markdown/HTML/图像 caption
权限隔离	PG RLS 行级安全 + 应用层双校验
引用溯源	100% 回答附 chunk_id + 页码

真实案例:某律所接入 IHUI AI 知识库,把 3000 份合同入库。律师问"竞业协议补偿标准",AI 回答 + 引用具体合同条款 + 页码,5 秒搞定。原来律师翻合同平均 25 分钟/次。

五、踩坑总结

坑 1:embedding 维度选错

一开始用 text-embedding-3-small(1536 维),后来想换 bge-m3(1024 维)省存储,发现维度不一致要重建索引。经验:一开始就评估好,后期换 embedding 等于重建知识库。

坑 2:切片太小召回发散

chunk_size=128 召回率高但 LLM 拿到的上下文太碎,回答支离破碎。我们最终用 512 token + 64 overlap,平衡召回和上下文完整。

坑 3:HNSW vs IVFFlat

HNSW 召回率高(95%+)但内存占用大;IVFFlat 省内存但召回率掉到 85%。100 万 chunks 内 HNSW 更优,千万级以上才考虑 IVFFlat。

坑 4:中文分词

pg_trgm 对中文不友好,我们把 jieba 集成到应用层做关键词提取,再喂给 SQL ILIKE。纯数据库层方案要上 pg_bigm(需要自己编译 PG 扩展)。

坑 5:权限过滤晚了一步

最初在 Python 层做权限过滤,有次 bug 把内部 HR 文档检索出来给普通员工看了。教训:权限必须在 SQL 层(RLS)兜底,应用层校验只是优化性能。

六、结语

企业级 RAG 的核心是:

切片智能:按 Markdown 标题 + 递归字符,表格/代码不切,512/64 是甜点。
三路融合:向量(语义)+ 全文(模糊)+ 关键词(精确),RRF 融合召回率 94%。
Rerank 兜底:bge-reranker-v2-m3 把 top-5 准确率从 81% 拉到 94%。
多模态:图表用 Vision LLM 生成 caption 入库,检索时一视同仁。
权限隔离:PG RLS 数据库层兜底 + 应用层校验,双保险。
引用溯源:每段回答附 chunk_id + 页码,员工敢用、法务敢批。

这套架构让 IHUI AI 的知识库在企业场景落地:律所、咨询公司、SaaS 厂商都在用。如果你也在做 RAG,强烈建议从第一天就用 pgvector + 混合检索——纯向量方案在生产环境必撞墙。

关于 IHUI AI

IHUI AI 是一站式 8 端全栈 AI 操作系统,Apache 2.0 开源。

🌐 官网:https://ihui.ai
💻 GitHub:https://github.com/IHUI-INF-AI/IHUI-AI(Star 支持一下 ⭐)
📦 8 端同源:Web / API / CLI / Desktop / Extension / Mobile / Miniapp
🤖 176 模型:OpenAI / Claude / Gemini / 通义 / DeepSeek / 智谱 / 文心 / 豆包 / Kimi / Ollama
💰 定价:Free / Pro ¥49/月 / Team ¥199/人/月 / Enterprise ¥2999/月起

5 分钟 Fork 到上线,替代 ChatGPT Team + Claude Code + Notion AI,月省 $60+。

This article was originally published on the IHUI AI Blog. Follow us on GitHub for more AI engineering content.