使用OpenAI Assistants API构建AI客户支持代理（2026版）

OpenAI Assistants API提供了托管状态、内置工具（文件搜索、代码解释器）和持久化线程——消除了构建AI代理时的大部分基础设施复杂性。本教程将构建一个完整的客户支持系统。

我们要构建什么

一个客户支持AI，能够：

从知识库（FAQ、文档）中回答问题

通过自定义函数查询订单状态

生成支持工单

跨会话保持对话上下文

环境准备

python
from openai import OpenAI
import json
import time
client = OpenAI()

第一步：创建知识库

上传你的文档作为文件：

python
上传FAQ和文档文件
files = []
for filename in ["faq.pdf", "product-docs.pdf", "return-policy.txt"]:
    with open(filename, "rb") as f:
        uploaded_file = client.files.create(
            file=f,
            purpose="assistants"
        )
        files.append(uploaded_file.id)
        print(f"已上传 {filename}: {uploaded_file.id}")
从文件创建向量存储
vector_store = client.beta.vector_stores.create(
    name="支持知识库",
    file_ids=files
)
等待处理完成
while vector_store.status == "in_progress":
    time.sleep(2)
    vector_store = client.beta.vector_stores.retrieve(vector_store.id)
print(f"向量存储就绪: {vector_store.id}")

第二步：创建助手

python
assistant = client.beta.assistants.create(
    name="TechCorp 支持代理",
    instructions="""你是一位TechCorp的客户支持代理。
指导原则：
保持友好、专业和同理心
在回答前始终搜索知识库
对于订单查询，使用get_order_status函数
如果无法解决问题，主动提出创建支持工单
绝不编造你不知道的信息
回复要简洁但完整""",
    model="gpt-4o",
    tools=[
        {"type": "file_search"},  # 对上传文件进行RAG
        {
            "type": "function",
            "function": {
                "name": "get_order_status",
                "description": "检索订单状态和物流信息",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "order_id": {
                            "type": "string",
                            "description": "订单ID（例如 ORD-12345）"
                        }
                    },
                    "required": ["order_id"]
                }
            }
        },
        {
            "type": "function",
            "function": {
                "name": "create_support_ticket",
                "description": "为未解决的问题创建支持工单",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "subject": {"type": "string"},
                        "description": {"type": "string"},
                        "priority": {
                            "type": "string",
                            "enum": ["low", "medium", "high", "urgent"]
                        }
                    },
                    "required": ["subject", "description", "priority"]
                }
            }
        }
    ],
    tool_resources={
        "file_search": {
            "vector_store_ids": [vector_store.id]
        }
    }
)
print(f"助手已创建: {assistant.id}")

第三步：实现工具处理函数

python
def handle_tool_call(tool_name: str, args: dict) -> str:
    if tool_name == "get_order_status":
        # 连接到实际的订单管理系统
        order_id = args["order_id"]
        # 模拟响应：
        orders_db = {
            "ORD-12345": {"status": "已发货", "tracking": "1Z999AA10123456784", "eta": "2026-01-15"},
            "ORD-67890": {"status": "处理中", "tracking": None, "eta": "2026-01-18"}
        }
        
        if order_id in orders_db:
            order = orders_db[order_id]
            return json.dumps(order)
        else:
            return json.dumps({"error": f"订单 {order_id} 未找到"})
    
    elif tool_name == "create_support_ticket":
        ticket_id = f"TICKET-{int(time.time())}"
        # 保存到你的工单系统（Zendesk、Jira等）
        print(f"创建工单: {args['subject']} | 优先级: {args['priority']}")
        return json.dumps({"ticket_id": ticket_id, "status": "已创建"})
    
    return json.dumps({"error": "未知工具"})

第四步：构建对话循环

python
def run_support_conversation(user_id: str, user_message: str, thread_id: str = None):
    # 创建或检索线程（线程持久化对话历史）
    if thread_id is None:
        thread = client.beta.threads.create()
        thread_id = thread.id
        print(f"新对话: {thread_id}")
    
    # 将用户消息添加到线程
    client.beta.threads.messages.create(
        thread_id=thread_id,
        role="user",
        content=user_message
    )
    
    # 运行助手
    run = client.beta.threads.runs.create(
        thread_id=thread_id,
        assistant_id=assistant.id
    )
    
    # 轮询直到完成
    while True:
        run = client.beta.threads.runs.retrieve(
            thread_id=thread_id,
            run_id=run.id
        )
        
        if run.status == "completed":
            break
        
        elif run.status == "requires_action":
            # 处理工具调用
            tool_outputs = []
            for tool_call in run.required_action.submit_tool_outputs.tool_calls:
                args = json.loads(tool_call.function.arguments)
                output = handle_tool_call(tool_call.function.name, args)
                tool_outputs.append({
                    "tool_call_id": tool_call.id,
                    "output": output
                })
            
            run = client.beta.threads.runs.submit_tool_outputs(
                thread_id=thread_id,
                run_id=run.id,
                tool_outputs=tool_outputs
            )
        
        elif run.status in ["failed", "cancelled", "expired"]:
            raise Exception(f"运行失败，状态: {run.status}")
        
        time.sleep(0.5)
    
    # 获取助手的回复
    messages = client.beta.threads.messages.list(thread_id=thread_id, order="desc", limit=1)
    response_text = messages.data[0].content[0].text.value
    
    return {"response": response_text, "thread_id": thread_id}
示例用法
result = run_support_conversation(
    user_id="user-123",
    user_message="我的订单ORD-12345状态如何？"
)
print(result["response"])
在同一个线程中继续对话
result2 = run_support_conversation(
    user_id="user-123",
    user_message="什么时候到货？",
    thread_id=result["thread_id"]  # 继续同一对话！
)
print(result2["response"])

第五步：部署为REST API

python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import uvicorn
app = FastAPI()
class ChatRequest(BaseModel):
    user_id: str
    message: str
    thread_id: Optional[str] = None
class ChatResponse(BaseModel):
    response: str
    thread_id: str
@app.post("/support/chat", response_model=ChatResponse)
async def support_chat(request: ChatRequest):
    try:
        result = run_support_conversation(
            user_id=request.user_id,
            user_message=request.message,
            thread_id=request.thread_id
        )
        return ChatResponse(**result)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

生产环境注意事项

线程管理

python
按用户会话存储线程ID
from redis import Redis
redis = Redis(host="localhost", port=6379, decode_responses=True)
def get_or_create_thread(user_id: str, session_id: str) -> str:
    key = f"thread:{user_id}:{session_id}"
    thread_id = redis.get(key)
    
    if not thread_id:
        thread = client.beta.threads.create()
        redis.setex(key, 3600, thread.id)  # 1小时后过期
        return thread.id
    
    return thread_id

成本估算

对于一个每天1000条消息的支持系统：

gpt-4o：约15-30美元/天

gpt-4o-mini：约1.50-3美元/天

对于简单的FAQ查询，考虑使用gpt-4o-mini，复杂问题则路由到gpt-4o。

结果与指标

使用OpenAI Assistants API进行客户支持的公司报告：

工单升级率降低65%

平均响应时间小于3秒

客户满意度评分87%

与传统支持人员相比成本降低70%

结论

OpenAI Assistants API使得构建生产质量的AI支持代理变得异常简单。借助内置的文件搜索、持久化线程和工具调用，你可以获得原本需要数周才能构建的能力。关键在于上传全面且组织良好的文档，并设计清晰、具体的工具函数。