大模型应用开发：使用 OpenAI Assistants 构建产品 AI 客服

2. 创建智能助手

使用 client.beta.assistants.create 创建客服实例。由于 Assistants API 处于 Beta 阶段，包名中包含 beta。

assistant = client.beta.assistants.create(
    name="空气净化器智能客服",
    description="24 小时为您服务，基于产品手册解答疑问",
    instructions="你是空气净化器公司的智能客服。请引用文件中的内容回答问题，表达要通俗易懂、尽量简短；若问题超出文件内容，请委婉拒绝。",
    model="gpt-3.5-turbo-1106",
    tools=[{"type": "retrieval"}],
    file_ids=[file_id]
)
assistant_id = assistant.id
print(f"Assistant created with ID: {assistant_id}")

参数说明：

• name：助手的显示名称。
• instructions：系统提示词，定义助手的角色和行为准则。这是控制输出质量的关键。
• model：指定使用的模型版本。目前推荐使用最新的 Turbo 版本以平衡成本与性能。
• tools：启用检索能力 (retrieval)，使助手能读取上传的文件。
• file_ids：关联已上传的知识库文件 ID。

3. 创建用户会话

每个用户的对话都需要独立的会话线程来维护上下文。

thread = client.beta.threads.create(
    metadata={
        "user_name": "张三",
        "user_id": "12345"
    }
)
thread_id = thread.id

4. 添加用户消息

将用户的问题添加到会话中。

message = client.beta.threads.messages.create(
    thread_id=thread_id,
    role="user",
    content="这款净化器支持哪些模式？"
)

5. 运行智能助手

触发助手处理当前会话的消息。

run = client.beta.threads.runs.create(
    thread_id=thread_id,
    assistant_id=assistant_id
)
run_id = run.id

6. 获取智能助手的回应

由于 Run 是异步处理的，我们需要轮询其状态直到完成。生产环境中建议使用指数退避策略（Exponential Backoff）以避免频繁请求。

import time

while run.status in ["queued", "in_progress"]:
    time.sleep(1) # 实际生产中建议增加随机延迟
    run = client.beta.threads.runs.retrieve(
        thread_id=thread_id,
        run_id=run_id
    )

if run.status == "completed":
    messages = client.beta.threads.messages.list(
        thread_id=thread_id,
        order="asc"
    )
    # 获取最后一条助手回复
    assistant_message = next(m for m in messages.data if m.role == "assistant")
    print(assistant_message.content[0].text)
elif run.status == "failed":
    print(f"Error: {run.last_error.message}")
else:
    print(f"Status: {run.status}")

状态说明：

• queued：任务排队中。
• in_progress：正在处理中。
• completed：处理成功。
• requires_action：需要调用函数（本例未涉及）。
• failed：运行失败。

7. 处理引用注解 (Annotations)

当助手引用知识库内容时，返回的消息中会包含 annotations。这允许我们在前端展示脚注，让用户点击查看原文来源。

def extract_message_content(message):
    text_content = message.content[0].text
    annotations = text_content.annotations
    
    citations = []
    value = text_content.value
    
    for index, annotation in enumerate(annotations):
        # 替换文本为脚注标记
        value = value.replace(annotation.text, f' [{index}]')
        
        # 收集引用信息
        if hasattr(annotation, 'file_citation'):
            cited_file = client.files.retrieve(annotation.file_citation.file_id)
            citations.append(f'[{index}] 引用自：{cited_file.filename}')
        elif hasattr(annotation, 'file_path'):
            cited_file = client.files.retrieve(annotation.file_path.file_id)
            citations.append(f'[{index}] 下载源文件：{cited_file.filename}')
    
    return value + '\n\n'.join(citations)

最佳实践与注意事项

1. API 密钥安全：永远不要在前端代码或公开仓库中暴露 API Key。应在服务端通过环境变量管理。
2. 费用控制：Retrieval 功能会增加 Token 消耗。建议定期清理不再需要的文件和 Thread。
3. 错误处理：网络波动可能导致 failed 状态。建议实现重试机制，针对 rate_limit_exceeded 等特定错误进行指数退避。
4. 内容合规：虽然 Retrieval 限制了回答范围，但仍需监控输出内容，防止敏感信息泄露或不当言论。
5. 模型选择：对于复杂的逻辑推理，可考虑升级到 GPT-4 系列模型，但需注意成本差异。

总结

通过 OpenAI Assistants API，开发者可以快速构建具备领域知识的智能客服，而无需从头训练模型。本文展示了从文件上传、助手配置到会话管理和响应获取的完整流程。借助检索增强生成（RAG）技术，企业能够更可靠地将大模型能力落地到具体业务场景中，提升用户体验的同时降低运营成本。