最新下载
热门教程
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
从脚本到平台：搭建可配置的 AI Agent 工厂
时间：2026-09-18 16:34:01 编辑：袖梨 来源：一聚教程网
用几段 Python 代码实现单个 Agent 并不困难，真正棘手的是让它持续服务多人，并支持频繁调整指令、切换模型和扩展工具。当智能体数量不断增加，硬编码脚本会迅速带来维护与协作负担。接下来将从平台化需求出发，拆解一个可配置、可编排且便于扩展的 Agent 系统。
从 0 到 1 搭建你的 AI Agent 平台：当 Agent 有了工厂，人人都能造同事
前端 AI Skill 体系 · 第 7 篇（Agent 平台篇）
上一篇，我们用 100 行代码造了一个 Research Agent，又用 200 行搭了一个 PM + Coder + Reviewer 三人团队。
跑通之后，我兴奋了大概两天。
第三天，问题来了。
同事问我："你这个 Agent 挺好用，能不能帮我也搞一个？"我说行，然后花了一下午帮他改
agent.py里的 instructions、换工具、调参数。第四天，另一个同事也来了。第五天，产品经理说："能不能让非技术人员自己配？"
我盯着终端里那个
python main.py，突然意识到一件事：我造了 3 个好同事，但我没有工厂。
每来一个需求，我就得手写一遍。改一个 Agent 的指令，就得打开编辑器、改代码、重启。加一个工具，就得写一个
@function_tool、注册、测试。这不是平台。这是手工作坊。
今天这篇，我们把它变成工厂。
一、为什么需要 Agent 平台（vs 手写脚本）
手写脚本的天花板
上一篇的 3 人团队，代不大，但维护成本不低：
| 操作 | 手写脚本 | 平台化 |
|---|---|---|
| 新建一个 Agent | 写 agent.py，定义 instructions、绑定 tools | 填表单，选模型，拖工具 |
| 修改 Agent 指令 | 改代码 → 重启进程 | 改配置 → 热更新 |
| 换一个模型 | 改 model="qwen-plus" → 重启 | 下拉框切换 |
| 加一个工具 | 写 @function_tool → 注册 → 重启 | 上传 / 从市场安装 |
| 给非技术人员用 | ❌ 不可能 | ✅ Web UI |
| 多 Agent 协作编排 | 手写 Handoff 代码 | 可视化拖拽 / YAML |
| 查看运行历史 | print 日志 | 结构化 Tracing 面板 |
手写脚本适合"我自己用"。平台适合"别人也能用"。
什么时候该做平台？
不是所有人都需要。我的判断标准：
| 信号 | 说明 |
|---|---|
| 第 3 个人来要 Agent | 你开始变成"Agent 客服" |
| 每周改 3 次以上 instructions | 配置和代码该分离了 |
| 需要非技术人员操作 | 终端不是所有人的朋友 |
| Agent 数量 > 5 | 手写 Handoff 开始混乱 |
| 需要审计/回溯 | "上周那个 Agent 为什么删了我的文件？" |
如果你只有 1-2 个 Agent、自己用、不怎么改——别做平台。 手写脚本够了。
但如果你命中了 3 条以上，继续往下看。
我们要造什么？
一句话：一个 Web 应用，让任何人都能通过浏览器创建、配置、编排、运行 AI Agent。
核心能力：
┌─────────────────────────────────────────────────────┐
│ Agent Platform │
│ │
│ ? 模型管理 多模型接入/切换/成本坚控 │
│ ? Skill 管理 上传/启用/禁用/版本控制 │
│ ? Agent 配置 角色/指令/工具绑定/模型选择 │
│ ? Agent 编排 可视化拖拽 or YAML 定义协作流程 │
│ ? 用户记忆 短期/长期/向量检索/跨会话持久化 │
│ ? MCP 插件 工具市场/第三方集成/自定义工具 │
│ │
│ 前端：React + Next.js（Web UI） │
│ 后端：Python FastAPI（Agent 运行时） │
│ 模型：OpenAI / 百炼 / Ollama（可切换） │
└─────────────────────────────────────────────────────┘
二、架构设计
技术选型：为什么是 React + Python
先说结论，再说为什么。
| 层 | 选型 | 为什么 |
|---|---|---|
| 前端 | React + Next.js + Tailwind CSS | 前端同学最熟、SSR 首屏快、组件生态丰富 |
| 后端 | Python FastAPI | Agent 生态全在 Python（OpenAI Agents SDK / LangChain / CrewAI） |
| 数据库 | SQLite → PostgreSQL | 个人用 SQLite 够了，多人用再切 PG |
| 向量库 | ChromaDB（本地）→ Qdrant（生产） | 记忆检索用，本地零依赖 |
| 模型层 | LiteLLM 统一代理 | 一套代码接 100+ 模型 |
| 部署 | Docker Compose | 一键起前端 + 后端 + 数据库 |
为什么不用 Node.js 做后端？
因为 Agent 运行时在 Python。OpenAI Agents SDK、LangChain、CrewAI、AutoGen——全是 Python。用 Node 做后端意味着你得通过 HTTP 调 Python 服务，多一层网络开销和调试成本。
为什么不用 Next.js 全栈？
Next.js 的 API Routes 跑 Python Agent 不现实。前后端分离，各司其职：Next.js 管 UI，FastAPI 管 Agent 运行时。
整体架构图
┌─────────────────────────────────────────────────────────────┐
│ 用户浏览器 │
│ ┌───────────────────────────────────────────────────────┐ │
│ │ Next.js (React + Tailwind) │ │
│ │ │ │
│ │ /agents Agent 列表 + 创建/编辑 │ │
│ │ /agents/:id Agent 详情 + 对话测试 │ │
│ │ /skills Skill 管理（上传/启用/禁用） │ │
│ │ /models 模型管理（接入/切换/测试） │ │
│ │ /workflows 编排画布（拖拽 or YAML） │ │
│ │ /memory 记忆管理（查看/搜索/清理） │ │
│ │ /plugins MCP 插件市场 │ │
│ │ /settings 全局设置 │ │
│ └───────────────────────┬───────────────────────────────┘ │
└──────────────────────────┼──────────────────────────────────┘
│ REST / WebSocket
┌──────────────────────────┼──────────────────────────────────┐
│ FastAPI (Python) │
│ │ │
│ ┌──────────┐ ┌────────┴───┐ ┌──────────┐ ┌──────────┐ │
│ │ Agent │ │ Workflow │ │ Memory │ │ Plugin │ │
│ │ Service │ │ Engine │ │ Service │ │ Registry │ │
│ │ │ │ │ │ │ │ │ │
│ │ CRUD │ │ Handoff │ │ 短期记忆 │ │ MCP 协议 │ │
│ │ 运行 │ │ 编排 │ │ 长期记忆 │ │ 工具市场 │ │
│ │ Tracing │ │ 重试 │ │ 向量检索 │ │ 自定义 │ │
│ └────┬─────┘ └────────────┘ └────┬─────┘ └──────────┘ │
│ │ │ │
│ ┌────┴──────────────────────────────┴─────┐ │
│ │ LiteLLM 统一模型代理 │ │
│ │ OpenAI / 百炼 / Claude / Ollama / ... │ │
│ └─────────────────────────────────────────┘ │
│ │
│ ┌──────────┐ ┌──────────┐ │
│ │ SQLite/PG│ │ ChromaDB │ │
│ │ 配置存储 │ │ 向量存储 │ │
│ └──────────┘ └──────────┘ │
└──────────────────────────────────────────────────────────────┘
项目目录结构
agent-platform/
│
├── frontend/ # Next.js 前端
│ ├── app/
│ │ ├── agents/ # Agent 管理页
│ │ │ ├── page.tsx # 列表
│ │ │ ├── new/page.tsx # 创建
│ │ │ └── [id]/page.tsx # 详情 + 对话
│ │ ├── skills/page.tsx # Skill 管理
│ │ ├── models/page.tsx # 模型管理
│ │ ├── workflows/page.tsx # 编排画布
│ │ ├── memory/page.tsx # 记忆管理
│ │ ├── plugins/page.tsx # MCP 插件
│ │ └── layout.tsx # 全局布局（侧边栏导航）
│ ├── components/ # 通用组件
│ │ ├── AgentCard.tsx
│ │ ├── ChatPanel.tsx
│ │ ├── WorkflowCanvas.tsx
│ │ └── ModelSelector.tsx
│ ├── lib/
│ │ └── api.ts # API 客户端
│ ├── package.json
│ └── tailwind.config.ts
│
├── backend/ # FastAPI 后端
│ ├── app/
│ │ ├── main.py # FastAPI 入口
│ │ ├── routers/
│ │ │ ├── agents.py # Agent CRUD + 运行
│ │ │ ├── skills.py # Skill 管理
│ │ │ ├── models.py # 模型管理
│ │ │ ├── workflows.py # 编排引擎
│ │ │ ├── memory.py # 记忆服务
│ │ │ └── plugins.py # MCP 插件
│ │ ├── services/
│ │ │ ├── agent_runner.py # Agent 运行时（核心）
│ │ │ ├── workflow_engine.py
│ │ │ ├── memory_store.py
│ │ │ └── model_proxy.py # LiteLLM 封装
│ │ ├── models/ # 数据模型（Pydantic）
│ │ │ ├── agent.py
│ │ │ ├── skill.py
│ │ │ └── workflow.py
│ │ └── db/
│ │ ├── database.py # SQLite/PG 连接
│ │ └── migrations/
│ ├── requirements.txt
│ └── Dockerfile
│
├── docker-compose.yml # 一键部署
├── .env.example
└── README.md
数据模型：Agent 长什么样
在写任何 UI 之前，先定义清楚"一个 Agent 在数据库里是什么"：
# backend/app/models/agent.py
from pydantic import BaseModel
from typing import Optional
from enum import Enum
class AgentStatus(str, Enum):
DRAFT = "draft"
ACTIVE = "active"
ARCHIVED = "archived"
class AgentConfig(BaseModel):
"""Agent 的完整配置——这就是平台的核心数据结构"""
# 基本信息
id: str
name: str # "Research Assistant"
description: str # "能搜索、能写文件的调研助手"
avatar: Optional[str] = None # 头像 URL
status: AgentStatus = AgentStatus.DRAFT
# 模型配置
model: str = "qwen-plus" # 模型标识
temperature: float = 0.7
max_tokens: int = 4096
# 核心：指令（就是上一篇的 instructions）
instructions: str = ""
# 工具绑定
tools: list[str] = [] # ["web_search", "read_file", "write_file"]
# Skill 绑定（Agent 可以加载多个 Skill 作为知识补充）
skills: list[str] = [] # ["code-review-rules", "writing-style"]
# 运行参数
max_turns: int = 10 # 最大工具调用轮次
needs_approval_tools: list[str] = [] # 需要人工确认的工具
# 编排（如果是多 Agent 协作）
handoffs: list[str] = [] # 可以交接给哪些 Agent ID
# 记忆配置
memory_enabled: bool = True
memory_type: str = "sqlite" # sqlite / redis / vector
# 元数据
created_at: str
updated_at: str
created_by: str = "admin"
对比上一篇：上一篇的 Agent 定义是硬编码在 agent.py 里的。现在，它变成了数据库里的一条记录。创建 Agent = INSERT，修改指令 = UPDATE，运行 Agent = SELECT + 构建 Agent() 对象。
这就是"平台化"的本质：把代码里的配置变成数据。
三、核心模块搭建
6 个模块，按依赖顺序搭：模型管理 → Skill 管理 → Agent 配置 → Agent 编排 → 用户记忆 → MCP 插件。
3.1 模型管理：多模型接入/切换
问题：上一篇我们硬编码了 model="qwen-plus"。如果用户想用 GPT-4o、Claude、本地 Ollama 呢？
方案：用 LiteLLM 做统一代理，前端只管选模型名。
# backend/app/services/model_proxy.py
from litellm import acompletion
from typing import Optional
# 平台支持的模型注册表
MODEL_REGISTRY = {
# 百炼/通义千问（国内直连）
"qwen-plus": {
"provider": "dashscope",
"display_name": "通义千问 Plus",
"max_tokens": 131072,
"cost_per_1k_input": 0.004, # 元
"cost_per_1k_output": 0.012,
},
"qwen-turbo": {
"provider": "dashscope",
"display_name": "通义千问 Turbo",
"max_tokens": 131072,
"cost_per_1k_input": 0.002,
"cost_per_1k_output": 0.006,
},
# OpenAI
"gpt-4o": {
"provider": "openai",
"display_name": "GPT-4o",
"max_tokens": 128000,
"cost_per_1k_input": 0.0175,
"cost_per_1k_output": 0.07,
},
"gpt-4o-mini": {
"provider": "openai",
"display_name": "GPT-4o Mini",
"max_tokens": 128000,
"cost_per_1k_input": 0.001,
"cost_per_1k_output": 0.004,
},
# 本地 Ollama
"ollama/llama3.1:8b": {
"provider": "ollama",
"display_name": "Llama 3.1 8B（本地）",
"max_tokens": 8192,
"cost_per_1k_input": 0,
"cost_per_1k_output": 0,
},
}
async def ch@t_completion(
model: str,
messages: list[dict],
temperature: float = 0.7,
max_tokens: int = 4096,
) -> str:
"""统一模型调用入口——不管底层是百炼、OpenAI 还是 Ollama"""
response = await acompletion(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
)
return response.choices[0].message.content
async def test_model_connection(model: str) -> dict:
"""测试模型连通性——前端"测试连接"按钮调用"""
try:
result = await ch@t_completion(
model=model,
messages=[{"role": "user", "content": "回复'连接成功'四个字"}],
max_tokens=20,
)
return {"status": "ok", "response": result}
except Exception as error:
return {"status": "error", "message": str(error)}
# backend/app/routers/models.py
from fastapi import APIRouter
from app.services.model_proxy import MODEL_REGISTRY, test_model_connection
router = APIRouter(prefix="/api/models", tags=["models"])
@router.get("/")
async def list_models():
"""返回所有可用模型——前端下拉框的数据源"""
return [
{"id": model_id, **config}
for model_id, config in MODEL_REGISTRY.items()
]
@router.post("/{model_id}/test")
async def test_connection(model_id: str):
"""测试模型连通性"""
if model_id not in MODEL_REGISTRY:
return {"status": "error", "message": f"未知模型：{model_id}"}
return await test_model_connection(model_id)
前端交互：模型管理页就是一个卡片列表，每张卡片显示模型名、提供商、价格、连通状态。点"测试连接"实时验证。创建 Agent 时，模型选择就是一个下拉框。
3.2 Skill 管理：上传/启用/禁用
问题：上一篇的 Skill 是放在项目目录里的 .md 文件。平台化之后，Skill 变成数据库记录，支持动态加载。
方案：Skill = 一段 Markdown 指令 + 元数据。Agent 运行时，把绑定的 Skill 内容拼接到 instructions 里。
# backend/app/models/skill.py
from pydantic import BaseModel
from typing import Optional
class SkillConfig(BaseModel):
id: str
name: str # "代码审查规范"
description: str # "定义代码审查的维度和标准"
content: str # Markdown 正文（就是 SKILL.md 的内容）
version: str = "1.0.0"
enabled: bool = True
tags: list[str] = [] # ["code-review", "frontend"]
created_at: str
updated_at: str
# backend/app/routers/skills.py
from fastapi import APIRouter, UploadFile, File
from app.db.database import get_db
router = APIRouter(prefix="/api/skills", tags=["skills"])
@router.get("/")
async def list_skills(enabled_only: bool = False):
"""列出所有 Skill"""
db = get_db()
query = "SELECT * FROM skills"
if enabled_only:
query += " WHERE enabled = 1"
query += " ORDER BY updated_at DESC"
return db.execute(query).fetchall()
@router.post("/upload")
async def upload_skill(file: UploadFile = File(...)):
"""上传 .md 文件创建 Skill——兼容上一篇的 SKILL.md 格式"""
content = (await file.read()).decode("utf-8")
# 从 Markdown 中提取 name 和 description
name = file.filename.replace(".md", "")
description = ""
for line in content.split("n"):
if line.startswith("# "):
name = line[2:].strip()
elif line.startswith("> ") and not description:
description = line[2:].strip()
db = get_db()
db.execute(
"INSERT INTO skills (name, description, content, enabled) VALUES (?, ?, ?, 1)",
(name, description, content),
)
db.commit()
return {"status": "ok", "name": name}
@router.patch("/{skill_id}/toggle")
async def toggle_skill(skill_id: str):
"""启用/禁用 Skill——不影响已绑定的 Agent，下次运行时生效"""
db = get_db()
db.execute("UPDATE skills SET enabled = NOT enabled WHERE id = ?", (skill_id,))
db.commit()
return {"status": "ok"}
Skill 和 Agent 的关系：Agent 的 instructions 是"你是谁、怎么工作"，Skill 是"你要遵守的规范"。运行时拼接：
# backend/app/services/agent_runner.py（片段）
def build_instructions(agent_config, skills: list[str]) -> str:
"""把 Agent 指令 + 绑定的 Skill 内容拼成完整 instructions"""
parts = [agent_config.instructions]
for skill_content in skills:
parts.append(f"nn---n## 附加规范n{skill_content}")
return "n".join(parts)
3.3 Agent 配置：角色/指令/工具绑定
这是平台的核心页面。用户通过表单创建一个 Agent，而不是写代码。
# backend/app/routers/agents.py
from fastapi import APIRouter
from agents import Agent, Runner, function_tool, set_default_openai_api
from app.models.agent import AgentConfig
from app.services.agent_runner import build_instructions, get_tools_by_names
from app.db.database import get_db
router = APIRouter(prefix="/api/agents", tags=["agents"])
@router.post("/")
async def create_agent(config: AgentConfig):
"""创建 Agent——前端表单提交"""
db = get_db()
db.execute(
"""INSERT INTO agents
(id, name, description, model, instructions, tools, skills,
max_turns, temperature, status)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
config.id, config.name, config.description,
config.model, config.instructions,
",".join(config.tools), ",".join(config.skills),
config.max_turns, config.temperature, "draft",
),
)
db.commit()
return {"status": "ok", "id": config.id}
@router.post("/{agent_id}/run")
async def run_agent(agent_id: str, user_input: str):
"""运行 Agent——前端对话面板调用"""
db = get_db()
row = db.execute("SELECT * FROM agents WHERE id = ?", (agent_id,)).fetchone()
if not row:
return {"error": "Agent 不存在"}
# 从数据库配置 → 构建 Agent 对象
skills_content = load_skills(row["skills"])
instructions = build_instructions(row, skills_content)
tools = get_tools_by_names(row["tools"].split(","))
agent = Agent(
name=row["name"],
model=row["model"],
instructions=instructions,
tools=tools,
)
# 百炼需要这两行（和上一篇一样）
if "qwen" in row["model"]:
import os
os.environ["OPENAI_AGENTS_DISABLE_TRACING"] = "true"
set_default_openai_api("ch@t_completions")
result = await Runner.run(
agent,
input=user_input,
max_turns=row["max_turns"],
)
return {"output": result.final_output}
前端表单长什么样？
┌─────────────────────────────────────────────────┐
│ 创建 Agent │
│ │
│ 名称： [Research Assistant ] │
│ 描述： [能搜索、能写文件的调研助手 ] │
│ │
│ 模型： [通义千问 Plus (qwen-plus) ▼] │
│ 温度： [0.7 ──────●────── ] │
│ │
│ 指令（System Prompt）： │
│ ┌─────────────────────────────────────────┐ │
│ │ 你是一个研究助理。 │ │
│ │ │ │
│ │ ## 工作流程 │ │
│ │ 1. 理解研究问题，拆解为子问题 │ │
│ │ 2. 使用 web_search 搜索 │ │
│ │ ... │ │
│ └─────────────────────────────────────────┘ │
│ │
│ 工具： ☑ web_search ☑ read_file ☑ write_file│
│ ☐ code_exec ☐ git_ops │
│ │
│ Skill： ☑ 代码审查规范 ☐ 写作风格指南 │
│ │
│ 最大轮次：[10] 需确认工具：[delete_file] │
│ │
│ [ 保存草稿 ] [ 保存并测试 ] │
└─────────────────────────────────────────────────┘
关键设计决策：指令编辑用纯文本 textarea，不用富文本编辑器。因为 instructions 本质是给 LLM 看的 Markdown，富文本反而添乱。
3.4 Agent 编排：可视化拖拽 or YAML
问题：上一篇的 Handoff 是硬编码的：
pm_agent.handoffs = [handoff(coder_agent)]
coder_agent.handoffs = [handoff(reviewer_agent)]
平台化之后，编排关系存在数据库里，支持两种编辑方式。
方案 A：YAML 定义（推荐起步）
# workflow: dev-team
name: 开发团队
description: PM → Coder → Reviewer 三人协作
entry_agent: pm-agent-id
agents:
- id: pm-agent-id
role: PM
handoffs: [coder-agent-id]
- id: coder-agent-id
role: Coder
handoffs: [reviewer-agent-id]
- id: reviewer-agent-id
role: Reviewer
handoffs: [coder-agent-id] # 打回重写
retry:
max_attempts: 2
on_failure: human_escalation
# backend/app/services/workflow_engine.py
import yaml
from agents import Agent, Runner, handoff
async def run_workflow(yaml_content: str, user_input: str) -> str:
"""解析 YAML 编排 → 构建 Handoff 链 → 运行"""
config = yaml.safe_load(yaml_content)
# 从数据库加载每个 Agent 的配置，构建 Agent 对象
agent_map = {}
for agent_def in config["agents"]:
agent_config = load_agent_from_db(agent_def["id"])
agent_map[agent_def["id"]] = build_agent(agent_config)
# 建立 Handoff 关系
for agent_def in config["agents"]:
agent = agent_map[agent_def["id"]]
agent.handoffs = [
handoff(agent_map[target_id])
for target_id in agent_def.get("handoffs", [])
]
# 从入口 Agent 开始运行
entry_agent = agent_map[config["entry_agent"]]
result = await Runner.run(
entry_agent,
input=user_input,
max_turns=20,
)
return result.final_output
方案 B：可视化拖拽（进阶）
用 React Flow 做拖拽画布：
┌──────────────────────────────────────────────────┐
│ 编排画布 │
│ │
│ ┌────────┐ ┌────────┐ ┌────────┐│
│ │ PM │────────▶│ Coder │────────▶│Reviewer││
│ │ 需求分析│ │ 编码实现│ │ 代码审查││
│ └────────┘ └────────┘ └───┬────┘│
│ ▲ │ │
│ │ 打回重写 │ │
│ └─────────────────┘ │
│ │
│ 拖拽节点 = 选 Agent │
│ 连线 = Handoff 关系 │
│ 双击节点 = 编辑 Agent 配置 │
│ 导出 = 生成 YAML │
└──────────────────────────────────────────────────┘
我的建议：先做 YAML，跑通后再加拖拽。拖拽是 UI 糖，YAML 是骨架。别本末倒置。
3.5 用户记忆：短期/长期/向量检索
问题：上一篇的 Agent 没有记忆。每次 Runner.run() 都是全新对话。用户说"继续上次的调研"，Agent 一脸懵。
方案：三层记忆架构。
┌─────────────────────────────────────────────┐
│ 记忆架构 │
│ │
│ 短期记忆（Working Memory） │
│ ├── 当前对话的上下文 │
│ ├── 存储：内存（Python dict） │
│ └── 生命周期：单次会话 │
│ │
│ 长期记忆（Long-term Memory） │
│ ├── 跨会话的事实和偏好 │
│ ├── "用户喜欢中文报告" │
│ ├── "用户的项目用 React + TypeScript" │
│ ├── 存储：SQLite / PostgreSQL │
│ └── 生命周期：永久（可手动清理） │
│ │
│ 语义记忆（Vector Memory） │
│ ├── 历史对话的向量化存储 │
│ ├── 支持"上次我们讨论过什么"类查询 │
│ ├── 存储：ChromaDB（本地）/ Qdrant（生产） │
│ └── 生命周期：可配置过期策略 │
└─────────────────────────────────────────────┘
# backend/app/services/memory_store.py
import sqlite3
import chromadb
from datetime import datetime
class MemoryStore:
"""Agent 的记忆管理器"""
def __init__(self, agent_id: str, user_id: str = "default"):
self.agent_id = agent_id
self.user_id = user_id
# 长期记忆 + 对话历史持久化：SQLite
self.db = sqlite3.connect("memory.db")
self.db.execute("""
CREATE TABLE IF NOT EXISTS long_term_memory (
id INTEGER PRIMARY KEY AUTOINCREMENT,
agent_id TEXT,
user_id TEXT,
content TEXT,
category TEXT DEFAULT 'fact',
created_at TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
self.db.execute("""
CREATE TABLE IF NOT EXISTS conversation_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
agent_id TEXT,
user_id TEXT,
conversation_id TEXT,
role TEXT,
content TEXT,
created_at TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
# 语义记忆：ChromaDB
self.chroma = chromadb.Client()
self.collection = self.chroma.get_or_create_collection(
name=f"agent_{agent_id}",
metadata={"hnsw:space": "cosine"},
)
def save_long_term(self, content: str, category: str = "fact"):
"""保存长期记忆——Agent 运行后自动提取关键信息"""
self.db.execute(
"INSERT INTO long_term_memory (agent_id, user_id, content, category) VALUES (?, ?, ?, ?)",
(self.agent_id, self.user_id, content, category),
)
self.db.commit()
def save_conversation(self, conversation_id: str, messages: list[dict]):
"""保存对话到向量库——支持语义检索"""
for i, msg in enumerate(messages):
self.collection.add(
documents=[msg["content"]],
ids=[f"{conversation_id}_{i}"],
metadatas=[{
"role": msg["role"],
"timestamp": datetime.now().isoformat(),
}],
)
def save_history(self, conversation_id: str, messages: list[dict]):
"""对话历史持久化——重启不丢，支持回看"""
for msg in messages:
self.db.execute(
"INSERT INTO conversation_history (agent_id, user_id, conversation_id, role, content) VALUES (?, ?, ?, ?, ?)",
(self.agent_id, self.user_id, conversation_id, msg["role"], msg["content"]),
)
self.db.commit()
def load_history(self, conversation_id: str) -> list[dict]:
"""加载某次对话的完整历史"""
rows = self.db.execute(
"SELECT role, content FROM conversation_history WHERE conversation_id = ? ORDER BY id",
(conversation_id,),
).fetchall()
return [{"role": row[0], "content": row[1]} for row in rows]
def list_conversations(self, limit: int = 20) -> list[dict]:
"""列出最近的对话——用于 UI 侧边栏展示"""
rows = self.db.execute(
"SELECT conversation_id, MIN(created_at) as started_at, COUNT(*) as msg_count FROM conversation_history WHERE agent_id = ? GROUP BY conversation_id ORDER BY started_at DESC LIMIT ?",
(self.agent_id, limit),
).fetchall()
return [{"conversation_id": r[0], "started_at": r[1], "message_count": r[2]} for r in rows]
def recall(self, query: str, top_k: int = 5) -> list[str]:
"""语义检索——"上次我们讨论过什么""""
results = self.collection.query(
query_texts=[query],
n_results=top_k,
)
return results["documents"][0] if results["documents"] else []
def get_context_for_prompt(self, user_input: str) -> str:
"""构建注入到 instructions 的记忆上下文"""
# 1. 取长期记忆
long_term = self.db.execute(
"SELECT content FROM long_term_memory WHERE agent_id = ? ORDER BY created_at DESC LIMIT 10",
(self.agent_id,),
).fetchall()
# 2. 语义检索相关历史
semantic = self.recall(user_input, top_k=3)
parts = []
if long_term:
facts = "n".join(f"- {row[0]}" for row in long_term)
parts.append(f"## 用户偏好和事实n{facts}")
if semantic:
history = "n".join(f"- {doc}" for doc in semantic)
parts.append(f"## 相关历史对话n{history}")
return "nn".join(parts) if parts else ""
运行时注入：
# agent_runner.py（片段）
async def run_agent_with_memory(agent_config, user_input: str):
memory = MemoryStore(agent_config.id)
# 把记忆上下文拼到 instructions 里
memory_context = memory.get_context_for_prompt(user_input)
full_instructions = agent_config.instructions
if memory_context:
full_instructions += f"nn---n{memory_context}"
agent = Agent(
name=agent_config.name,
model=agent_config.model,
instructions=full_instructions,
tools=get_tools_by_names(agent_config.tools),
)
result = await Runner.run(agent, input=user_input, max_turns=agent_config.max_turns)
# 运行后保存对话
messages = [
{"role": "user", "content": user_input},
{"role": "assistant", "content": result.final_output},
]
memory.save_conversation(conversation_id=result.run_id, messages=messages) # 向量库（语义检索）
memory.save_history(conversation_id=result.run_id, messages=messages) # SQLite（持久化回看）
return result.final_output
3.6 MCP 插件：工具市场
上一篇的工具是硬编码在 tools.py 里的。平台化之后，工具应该像插件一样可安装、可卸载。
MCP（Model Context Protocol） 是 Anthropic 提出的开放协议，2026 年已成为 Agent 工具生态的事实标准。平台的插件系统支持三种工具来源：内置工具（开箱即用）、MCP Server（社区生态，即插即用）、自定义工具（用户上传 Python 函数，沙箱执行）。
? 插件注册表（
PluginRegistry）、MCP Server 通信、安装/卸载 API 的完整实现 → 评论区扣「Agent平台」或私信获取
四、Web UI 实现
技术栈：Next.js 15 + React 19 + Tailwind CSS + shadcn/ui
为什么选这套？前端同学最熟、组件生态最丰富、AI 生成代码的质量最高（训练数据里 React 代码最多）。
前端包含 7 个页面（Agent 管理、Skill 管理、模型管理、编排画布、记忆管理、插件市场、全局设置），文章篇幅原因，这里只展示最核心的交互——对话测试面板，其余页面的完整代码在完整版中。
? 完整代码（含 Web UI 全部页面 + API Key 管理面板）→ 评论区扣「Agent平台」或私信获取
4.1 对话测试面板
创建完 Agent，最重要的交互是立刻测试。
// frontend/components/ChatPanel.tsx
"use client"
import { useState, useRef, useEffect } from "react"
import { api } from "@/lib/api"
interface Message {
role: "user" | "assistant"
content: string
}
export function ChatPanel({ agentId }: { agentId: string }) {
const [messages, setMessages] = useState<Message[]>([])
const [input, setInput] = useState("")
const [loading, setLoading] = useState(false)
const bottomRef = useRef<HTMLDivElement>(null)
useEffect(() => {
bottomRef.current?.scrollIntoView({ behavior: "smooth" })
}, [messages])
async function sendMessage() {
if (!input.trim() || loading) return
const userMessage = input.trim()
setInput("")
setMessages((prev) => [...prev, { role: "user", content: userMessage }])
setLoading(true)
try {
const result = await api.post(`/api/agents/${agentId}/run`, {
user_input: userMessage,
})
setMessages((prev) => [
...prev,
{ role: "assistant", content: result.output },
])
} catch (error) {
setMessages((prev) => [
...prev,
{ role: "assistant", content: `❌ 运行出错：${error}` },
])
} finally {
setLoading(false)
}
}
return (
<div className="flex h-[600px] flex-col rounded-xl border border-zinc-800 bg-zinc-900">
{/* 消息列表 */}
<div className="flex-1 space-y-4 overflow-y-auto p-4">
{messages.length === 0 && (
<p className="mt-20 text-center text-sm text-zinc-500">
发送消息开始测试你的 Agent ?
</p>
)}
{messages.map((msg, i) => (
<div
key={i}
className={`flex ${msg.role === "user" ? "justify-end" : "justify-start"}`}
>
<div
className={`max-w-[80%] rounded-lg px-4 py-2 text-sm ${
msg.role === "user"
? "bg-emerald-600 text-white"
: "bg-zinc-800 text-zinc-200"
}`}
>
<pre className="whitespace-pre-wrap font-sans">{msg.content}</pre>
</div>
</div>
))}
{loading && (
<div className="flex justify-start">
<div className="rounded-lg bg-zinc-800 px-4 py-2 text-sm text-zinc-400">
? 思考中...
</div>
</div>
)}
<div ref={bottomRef} />
</div>
{/* 输入框 */}
<div className="border-t border-zinc-800 p-4">
<div className="flex gap-2">
<input
value={input}
onChange={(e) => setInput(e.target.value)}
onKeyDown={(e) => e.key === "Enter" && sendMessage()}
placeholder="输入消息测试 Agent..."
className="flex-1 rounded-lg border border-zinc-700 bg-zinc-800 px-4 py-2 text-sm text-zinc-100 placeholder-zinc-500 focus:border-emerald-500 focus:outline-none"
/>
<button
onClick={sendMessage}
disabled={loading}
className="rounded-lg bg-emerald-600 px-4 py-2 text-sm font-medium hover:bg-emerald-500 disabled:opacity-50"
>
发送
</button>
</div>
</div>
</div>
)
}
其余页面（全局布局、Agent 列表、API 客户端、Skill 管理、模型管理、编排画布、记忆管理、插件市场、全局设置）的完整代码，篇幅原因不在这里展开。
? 完整代码（含 Web UI 全部 7 个页面 + API 客户端 + API Key 管理面板）→ 评论区扣「Agent平台」或私信获取
五、部署与商业化思考
5.1 一键部署：Docker Compose
# docker-compose.yml
version: "3.8"
services:
frontend:
build: ./frontend
ports:
- "3000:3000"
environment:
- NEXT_PUBLIC_API_URL=http://backend:8000
depends_on:
- backend
backend:
build: ./backend
ports:
- "8000:8000"
env_file:
- .env
volumes:
- ./data:/app/data # SQLite + ChromaDB 持久化
depends_on:
- ollama
ollama:
image: ollama/ollama:latest
ports:
- "11434:11434"
volumes:
- ollama_data:/root/.ollama
volumes:
ollama_data:
# 一键启动
docker compose up -d
# 访问
# 前端：http://localhost:3000
# 后端 API：http://localhost:8000/docs（FastAPI 自带 Swagger）
# Ollama：http://localhost:11434
5.2 商业化路径：从个人到企业
| 阶段 | 用户 | 功能 | 收费 |
|---|---|---|---|
| 开源版 | 个人开发者 | 单用户、SQLite、本地模型 | 免费 |
| Pro 版 | 小团队（5人） | 多用户、PostgreSQL、云端模型、Tracing | ¥99/月 |
| 企业版 | 企业 | SSO、RBAC、审计日志、私有部署、SLA | 按需报价 |
开源是获客手段，不是商业模式。 核心壁垒不在代码，在于：
- 模型调优经验（什么 instructions 效果好）
- Skill 生态（社区贡献的高质量 Skill）
- 工具集成深度（MCP 插件数量和质量）
5.3 成本估算
| 组件 | 个人用 | 团队用 |
|---|---|---|
| 服务器 | 本地 / 免费云 | 2C4G 云服务器 ~¥100/月 |
| 模型 API | 百炼免费额度 / Ollama 本地 | ~¥200-500/月 |
| 向量库 | ChromaDB 本地 | Qdrant Cloud ~$25/月 |
| 域名 + SSL | ~¥60/年 | 同左 |
| 总计 | ¥0 | ¥300-600/月 |
六、Q&A
Q1: 为什么不直接用 Dify / Coze / FastGPT？
好问题。它们确实能做很多本文提到的事。
| 维度 | Dify / Coze | 自建平台 |
|---|---|---|
| 上手速度 | 快，注册即用 | 慢，要搭环境 |
| 定制深度 | 受限于平台能力 | 完全可控 |
| 数据隐私 | 数据在别人服务器 | 数据在自己手里 |
| 学习价值 | 低（拖拖拽拽） | 高（理解 Agent 全链路） |
| 成本 | 免费额度有限 | 本地模型零成本 |
如果你只是想用 Agent，用 Dify。如果你想理解 Agent 怎么工作、想完全控制、想给团队定制——自建。
本文的目的不是"造一个比 Dify 好的产品"，是让你理解 Agent 平台的每一个齿轮是怎么转的。理解了，你用什么平台都能用得更好。
Q2: 前端用 Vue / Svelte 行不行？
行。本文选 React + Next.js 是因为：
- 前端同学最熟
- shadcn/ui 组件生态最丰富
- AI 生成 React 代码的质量最高（训练数据多）
但架构是前后端分离的，前端换成 Vue + Nuxt、Svelte + SvelteKit 都行，后端 API 不用改。
Q3: 后端为什么不用 Node.js？
因为 Agent 运行时在 Python。OpenAI Agents SDK、LangChain、CrewAI 全是 Python 生态。用 Node 做后端意味着：
- 要么通过 HTTP 调 Python 微服务（多一层网络）
- 要么用 Node 的 Agent SDK（生态远不如 Python）
Python 做 Agent 运行时，Node/React 做 UI。各用各的强项。
Q4: 多用户怎么做？
开源版是单用户。多用户需要加：
# 最简方案：JWT 认证
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer
security = HTTPBearer()
async def get_current_user(token=Depends(security)):
user = verify_jwt(token.credentials)
if not user:
raise HTTPException(status_code=401, detail="未授权")
return user
# 每个 Agent 加 owner_id 字段
# 每个查询加 WHERE owner_id = ?
别一开始就做多用户。先单用户跑通，有需求再加。
Q5: 安全怎么保证？
| 风险 | 对策 |
|---|---|
| Agent 执行危险操作 | needs_approval=True + 工具白名单 |
| 用户自定义工具注入 | 沙箱执行（Docker 容器 / gVisor） |
| API Key 泄露 | 后端存储，前端永远不接触 Key |
| Prompt 注入 | 输入 Guardrail + 输出 Guardrail |
| 数据隔离 | 每个用户的 Agent/记忆严格隔离 |
Q6: 这个平台能商用吗？
能，但要注意：
- 模型 API 的 ToS（Terms of Service）——百炼和 OpenAI 都允许商用，但要遵守用量限制
- 用户数据合规——如果涉及个人信息，需要隐私协议
- 开源协议——如果基于本文代码，MIT 协议，随便用
Q7: 下一步做什么？
| 优先级 | 功能 | 原因 |
|---|---|---|
✅ 已内置（3.5 记忆模块的 save_history / load_history） | ||
| P0 | Agent 版本管理 | 改坏了能回滚 |
| P1 | 流式输出（SSE） | 用户体验质的飞跃 |
| P1 | 多用户 + 权限 | 团队使用的前提 |
| P2 | 可视化编排（React Flow） | YAML 够用了，拖拽是锦上添花 |
| P2 | 插件市场（社区贡献） | 生态效应 |
| P3 | 移动端适配 | 有 Web 就行，别过早优化 |
速记表
| 想做什么 | 用什么 |
|---|---|
| 多模型切换 | LiteLLM 统一代理 |
| Skill 动态加载 | 数据库记录 + 运行时拼接 |
| Agent 配置化 | Pydantic 数据模型 + SQLite |
| 多 Agent 编排 | YAML 定义 Handoff 链 |
| 记忆持久化 | SQLite（长期）+ ChromaDB（语义） |
| 工具插件化 | MCP 协议 + 沙箱执行 |
| 一键部署 | Docker Compose |
| 非技术人员使用 | Web UI（Next.js） |
结语
上一篇，我们造了 3 个好同事。 这一篇，我们建了一座工厂。
第 5 篇：写一份岗位说明书（Skill）
第 6 篇：招一个活人（Agent）
第 7 篇（本篇）：建一座工厂（Platform）
从 30 行 Markdown 到 100 行 Python 到一整个 Web 应用——每一步都在回答同一个问题：
怎么让 AI 从"工具"变成"同事"，再变成"团队"？
工厂不是一天建成的。本文给你的是蓝图和地基：
- 6 个核心模块的数据结构和 API
- 前端关键页面的完整代码
- Docker 一键部署
- 从个人到企业的商业化路径
你不需要一次全做完。 先跑通"创建 Agent → 配置 → 对话测试"这条最短路径。一个周末就够了。
然后，每加一个模块，你的工厂就多一条生产线。
直到有一天，你的同事打开浏览器，自己拖了一个 Agent 出来，试了两句话，说："挺好用的。"
那一刻，你不再是一个写脚本的人。
你是一个造同事的人。
? 本文所有完整代码（含 Web UI 全部 7 个页面 + API Key 管理面板 + MCP 插件完整实现 + 环境搭建指南）→ 评论区扣「Agent平台」或私信获取
系列回顾：
- 第 1 篇：《基于四层Skill体系的前端团队AI提效实践》
- *第 2 篇：《别再凭感觉调 Prompt 了：8 个 Demo 带你把 Prompt 当代码管》
- 第 3 篇：《一份 AGENTS.md，让 AI 代码规范率从 60% 飙升到 95%》
- 第 4 篇：《你的 AI Skill 越多越蠢？Token 上下文爆炸的求生指南》
- 第 5 篇：《手把手书写你的第一个 AI Skill》
- 第 6 篇：《手把手书写你的第一个 AI Agent》
- 第 7 篇（本篇）：《从 0 到 1 搭建你的 AI Agent 平台》
- 下一篇：敬请期待
参考来源
| 来源 | 内容 |
|---|---|
| OpenAI Agents SDK | Agent / Handoff / Guardrails / Tracing |
| LiteLLM | 多模型统一代理（100+ 模型） |
| FastAPI | Python Web 框架（自动 Swagger） |
| Next.js 15 | React 全栈框架（SSR / App Router） |
| shadcn/ui | React 组件库 |
| React Flow | 可视化流程编排 |
| ChromaDB | 本地向量数据库 |
| MCP（Model Context Protocol） | Anthropic 提出的工具调用标准协议 |
| Dify / Coze / FastGPT | 现有 Agent 平台（对比参考） |
相关文章
- asp服务器如何搭建 09-18
- 用 CueCast MCP 搭建自动化测试 Agent 工作流 09-18
- 从脚本到平台：搭建可配置的 AI Agent 工厂 09-18
- 用 AI 规划十一行程：快速生成可共享的旅行网页 09-18
- chat.asp聊天程序的编写方法 09-18
- 构建支持多种输出形态的 Agent Chat 09-18