OpenAI Assistants API vs 自建 Agent 系统

概述

构建 AI Agent 应用时，团队面临一个根本性选择：使用 OpenAI Assistants API 这样的托管服务，还是基于 LangGraph/AutoGen 等框架自建 Agent 系统。

这不仅是技术选型问题，更是业务战略问题：托管服务降低开发成本但牺牲控制力，自建系统获得完全灵活性但承担全部运维复杂度。

本文从功能覆盖、架构差异、成本模型、适用场景四个维度进行系统对比。

Assistants API 架构

用户应用
    |
    v
OpenAI Assistants API
    |
    ├── Thread（对话线程）
    │   └── 自动管理消息历史和上下文窗口
    |
    ├── Run（执行引擎）
    │   ├── 模型推理
    │   ├── 工具调用循环
    │   └── 状态管理（queued -> in_progress -> completed）
    |
    ├── 内置工具
    │   ├── Code Interpreter（沙箱代码执行）
    │   ├── File Search（向量检索 RAG）
    │   └── Function Calling（自定义工具）
    |
    └── 存储
        ├── Vector Store（向量数据库）
        └── File Storage（文件存储）

核心 API 使用

from openai import OpenAI

client = OpenAI()

# 1. 创建 Assistant
assistant = client.beta.assistants.create(
    name="数据分析师",
    instructions="""你是一个数据分析专家。
使用 Code Interpreter 分析用户上传的数据文件。
生成可视化图表并给出分析结论。""",
    model="gpt-4o",
    tools=[
        {"type": "code_interpreter"},
        {"type": "file_search"},
        {
            "type": "function",
            "function": {
                "name": "get_company_info",
                "description": "获取公司基本信息",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "company_name": {"type": "string"},
                    },
                    "required": ["company_name"],
                },
            },
        },
    ],
)

# 2. 创建 Vector Store（用于 File Search）
vector_store = client.beta.vector_stores.create(name="knowledge_base")
client.beta.vector_stores.file_batches.upload_and_poll(
    vector_store_id=vector_store.id,
    files=[open("docs/manual.pdf", "rb")],
)

# 更新 Assistant 绑定 Vector Store
client.beta.assistants.update(
    assistant.id,
    tool_resources={"file_search": {"vector_store_ids": [vector_store.id]}},
)

# 3. 创建 Thread 并运行
thread = client.beta.threads.create()

# 上传文件到 Thread
file = client.files.create(file=open("sales_data.csv", "rb"), purpose="assistants")

client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="分析这份销售数据，找出月度趋势和异常值",
    attachments=[{"file_id": file.id, "tools": [{"type": "code_interpreter"}]}],
)

# 4. 创建 Run（执行）
run = client.beta.threads.runs.create_and_poll(
    thread_id=thread.id,
    assistant_id=assistant.id,
)

# 5. 处理工具调用
if run.status == "requires_action":
    tool_outputs = []
    for tool_call in run.required_action.submit_tool_outputs.tool_calls:
        if tool_call.function.name == "get_company_info":
            args = json.loads(tool_call.function.arguments)
            result = lookup_company(args["company_name"])
            tool_outputs.append({
                "tool_call_id": tool_call.id,
                "output": json.dumps(result),
            })

    run = client.beta.threads.runs.submit_tool_outputs_and_poll(
        thread_id=thread.id,
        run_id=run.id,
        tool_outputs=tool_outputs,
    )

# 6. 获取结果
messages = client.beta.threads.messages.list(thread_id=thread.id)
for msg in messages.data:
    if msg.role == "assistant":
        for block in msg.content:
            if block.type == "text":
                print(block.text.value)
            elif block.type == "image_file":
                # 下载生成的图表
                image_data = client.files.content(block.image_file.file_id)

流式执行

from openai import AssistantEventHandler

class MyEventHandler(AssistantEventHandler):
    def on_text_created(self, text):
        print("\nassistant > ", end="", flush=True)

    def on_text_delta(self, delta, snapshot):
        print(delta.value, end="", flush=True)

    def on_tool_call_created(self, tool_call):
        print(f"\n[Tool: {tool_call.type}]", flush=True)

    def on_tool_call_delta(self, delta, snapshot):
        if delta.type == "code_interpreter" and delta.code_interpreter.input:
            print(delta.code_interpreter.input, end="", flush=True)

with client.beta.threads.runs.stream(
    thread_id=thread.id,
    assistant_id=assistant.id,
    event_handler=MyEventHandler(),
) as stream:
    stream.until_done()

自建 Agent 系统架构

用户应用
    |
    v
自建 Agent 框架 (LangGraph / 自研)
    |
    ├── 对话管理
    │   ├── 消息存储（PostgreSQL / Redis）
    │   ├── 上下文窗口管理（摘要压缩 / 滑动窗口）
    │   └── 多轮对话状态机
    |
    ├── 推理引擎
    │   ├── 模型路由（OpenAI / Anthropic / 本地模型）
    │   ├── Prompt 管理（模板 / 版本控制）
    │   └── 输出解析 + 重试
    |
    ├── 工具系统
    │   ├── 工具注册中心
    │   ├── 权限控制
    │   ├── 执行沙箱
    │   └── 结果缓存
    |
    ├── RAG 管道
    │   ├── 文档解析（PDF / Word / HTML）
    │   ├── 分块策略
    │   ├── Embedding 生成
    │   ├── 向量数据库（Qdrant / Milvus）
    │   └── 检索 + 重排
    |
    └── 可观测性
        ├── 日志 + 追踪
        ├── 指标监控
        └── 成本追踪

核心实现示例

# 自建 Agent 核心循环（简化版）
class CustomAgent:
    def __init__(self, model_client, tools, rag_pipeline, memory):
        self.model = model_client
        self.tools = tools
        self.rag = rag_pipeline
        self.memory = memory

    async def run(self, user_message: str, session_id: str) -> str:
        # 1. 加载历史上下文
        history = await self.memory.load(session_id)

        # 2. RAG 检索
        relevant_docs = await self.rag.retrieve(user_message, top_k=5)
        context = "\n".join([doc.content for doc in relevant_docs])

        # 3. 构建 prompt
        messages = [
            {"role": "system", "content": self.system_prompt + f"\n\nContext:\n{context}"},
            *history,
            {"role": "user", "content": user_message},
        ]

        # 4. Agent 循环
        max_iterations = 10
        for i in range(max_iterations):
            response = await self.model.chat(messages, tools=self.tools.schemas)

            if not response.tool_calls:
                # 没有工具调用，返回最终结果
                await self.memory.save(session_id, messages + [response])
                return response.content

            # 执行工具调用
            messages.append(response)
            for tc in response.tool_calls:
                result = await self.tools.execute(tc.name, tc.args)
                messages.append({
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": json.dumps(result),
                })

        return "达到最大迭代次数"

功能对比

成本模型对比

Assistants API 成本

1. 模型推理成本（与直接 API 调用相同）
   - GPT-4o: $2.50 / 1M input tokens, $10.00 / 1M output tokens

2. Code Interpreter 会话成本
   - $0.03 / session（每个 Run 一个 session）

3. File Search 成本
   - Vector Store: $0.10 / GB / 天
   - 检索: 包含在 token 成本中

4. 文件存储
   - 包含在 Vector Store 费用中

估算（月度，中等用量）：
- 10,000 次对话 x 平均 2,000 tokens = 20M tokens
- Token 成本: ~$50 (input) + ~$100 (output) = $150
- Code Interpreter: 1,000 次 x $0.03 = $30
- Vector Store: 10GB x $0.10 x 30 = $30
- 月度总计: ~$210

自建系统成本

1. 模型推理（可选多提供商）
   - Claude Sonnet: $3.00 / 1M input, $15.00 / 1M output
   - GPT-4o-mini: $0.15 / 1M input, $0.60 / 1M output（成本敏感场景）
   - 本地模型: GPU 服务器成本

2. 基础设施
   - 向量数据库（Qdrant Cloud / 自托管）: $50-200/月
   - 应用服务器: $50-200/月
   - 数据库（PostgreSQL）: $20-100/月
   - Redis: $10-50/月
   - 监控（Datadog/Grafana）: $50-200/月

3. 开发与运维
   - 工程师时间: 1-3 人月开发 + 持续维护

估算（月度，中等用量）：
- 模型推理（混合路由）: ~$100
- 基础设施: ~$200-500
- 运维人力（分摊）: ~$500-1000
- 月度总计: ~$800-1,600

成本交叉点

              成本
               |
               |        自建系统（固定成本高，边际成本低）
               |       /
               |      /
               |     /
               |    /
               |   /        Assistants API（固定成本低，按量计费）
               |  / -------/
               | /  /
               |/ /
               |/
               +───────────────────────────> 使用量
                     交叉点 (~50K 请求/月)

大致规律：

• < 10K 请求/月：Assistants API 更经济
• 10K-50K 请求/月：取决于具体场景
• 50K 请求/月：自建系统开始有成本优势

选型决策

选 Assistants API 当

• 快速 MVP / 原型验证
• 团队缺少 AI 工程经验
• 主要使用 OpenAI 模型
• 对数据驻留没有严格要求
• 不需要多 Agent 协作
• 不需要精细的 RAG 控制
• 使用量适中（< 50K 请求/月）

选自建系统当

• 需要多模型路由（成本/性能/合规）
• 需要精细的 RAG 管道控制
• 需要多 Agent 协作
• 有数据驻留/合规要求
• 大规模使用（> 50K 请求/月）
• 需要深度自定义和可观测性
• 团队有 AI 工程能力

混合方案

# 实际上，很多团队采用混合方案
class HybridAgent:
    """混合方案：Assistants API 做重活，自建做轻量和路由"""

    def __init__(self):
        self.openai = OpenAI()
        self.anthropic = Anthropic()
        self.local_tools = LocalToolRegistry()

    async def route(self, request):
        # 需要代码执行 -> Assistants API
        if request.needs_code_execution:
            return await self._run_assistant(request)

        # 简单对话 -> 直接 API（成本更低）
        if request.complexity == "simple":
            return await self._direct_chat(request)

        # 敏感数据 -> 本地模型
        if request.contains_sensitive_data:
            return await self._local_inference(request)

        # 复杂多步任务 -> 自建 Agent
        return await self._custom_agent(request)

迁移路径

从 Assistants API 到自建

阶段 1: 并行运行
  - 新功能用自建系统
  - 旧功能继续用 Assistants API
  - 对比两个系统的效果

阶段 2: 核心能力替代
  - 自建 RAG 管道替代 File Search
  - 自建代码沙箱替代 Code Interpreter
  - 自建对话管理替代 Thread API

阶段 3: 完全迁移
  - 所有流量切到自建系统
  - 保留 OpenAI 作为模型提供商之一
  - Assistants API 作为备选方案

关键提醒

1. 不要过早自建：如果 Assistants API 能满足 80% 的需求，先用它跑起来
2. 不要过度依赖：把核心业务逻辑放在自己的代码中，而非 Assistants API 的 instructions
3. 保留切换能力：在应用层做一个 AgentInterface 抽象，底层可以切换实现
4. 监控成本：Assistants API 的 Code Interpreter 按 session 计费，频繁调用成本会快速上升

总结

核心建议：先用 Assistants API 验证产品想法，跑通商业模式后再根据实际瓶颈决定是否自建。

Maurice | [email protected]