共计 2575 个字符,预计需要花费 7 分钟才能阅读完成。
背景痛点
最近在帮朋友公司对接客服系统时,发现传统的关键词匹配方案经常答非所问。调研后发现 ChatGPT 的对话理解能力明显更强,但他们技术团队只有 2 人,需要快速实现以下目标:

- 5 秒内响应客户咨询
- 支持多轮对话上下文
- 零费用启动(前期免费额度足够)
技术选型对比
测试了两种接入方式,数据基于 100 次连续调用(单位:ms):
| 指标 | 官方 REST API | 第三方 WebSocket 库 |
|---|---|---|
| 平均延迟 | 320 | 210 |
| 99 分位延迟 | 810 | 450 |
| 错误率 | 0.3% | 1.8% |
结论:
– 需要稳定性的选官方 API
– 追求低延迟可尝试 WebSocket(注意重连机制)
核心实现
认证流程
sequenceDiagram
开发者 ->>+OpenAI: 1. POST /oauth/token
OpenAI-->>- 开发者: access_token(有效期 2h)
开发者 ->>+OpenAI: 2. 带 token 调用 API
OpenAI-->>- 开发者: 流式返回结果
Python 示例
import requests
from tenacity import retry, stop_after_attempt
class ChatGPT:
def __init__(self, api_key):
self.session = requests.Session()
self.session.headers.update({'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
})
@retry(stop=stop_after_attempt(3))
def ask(self, prompt):
try:
resp = self.session.post(
'https://api.openai.com/v1/chat/completions',
json={
'model': 'gpt-3.5-turbo',
'messages': [{'role': 'user', 'content': prompt}]
},
timeout=10
)
resp.raise_for_status()
return resp.json()['choices'][0]['message']['content']
except Exception as e:
print(f'调用失败: {str(e)}')
raise
# 测试用例
def test_chat():
bot = ChatGPT(os.getenv('OPENAI_KEY'))
assert '你好' in bot.ask('用中文打招呼')
Node.js 流式处理
const {OpenAI} = require('openai');
class ChatStream {constructor(apiKey) {this.client = new OpenAI({ apiKey});
}
async *askStream(prompt) {
const stream = await this.client.chat.completions.create({
model: 'gpt-3.5-turbo',
messages: [{role: 'user', content: prompt}],
stream: true
});
for await (const chunk of stream) {yield chunk.choices[0]?.delta?.content || '';
}
}
}
// 使用示例
(async () => {const chat = new ChatStream(process.env.OPENAI_KEY);
for await (const text of chat.askStream('讲个笑话')) {process.stdout.write(text);
}
})();
生产环境方案
限流控制
from ratelimit import limits, sleep_and_retry
class RateLimitedGPT(ChatGPT):
@sleep_and_retry
@limits(calls=3, period=1) # 每秒 3 次
def ask(self, prompt):
return super().ask(prompt)
敏感词过滤
推荐在返回结果后添加处理层:
function filterSensitive(text) {const blacklist = ['暴力', '政治'];
return blacklist.some(word => text.includes(word))
? '内容不符合规范'
: text;
}
常见问题
- 风控触发 :
- 同一 IP 避免高频登录
-
推荐复用 access_token
-
代码块转义 :
import markdown def render_response(text): return markdown.markdown(text.replace('```', '\n```'))
进阶应用
记忆对话上下文
使用 LangChain 的 Memory 模块:
from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory()
memory.save_context({'input': '推荐周末活动'},
{'output': '可以去公园野餐'}
)
print(memory.load_memory_variables({}))
监控指标
Prometheus 埋点示例:
from prometheus_client import Counter, Summary
REQUEST_TIME = Summary('chatgpt_latency', '请求耗时')
ERROR_COUNT = Counter('chatgpt_errors', '错误次数')
@REQUEST_TIME.time()
def monitored_ask(prompt):
try:
return ask(prompt)
except Exception:
ERROR_COUNT.inc()
raise
使用体验
实际测试发现官方 API 在晚高峰时段偶尔会出现 500 错误,通过添加指数退避的重试机制后稳定性明显提升。对于需要快速响应的场景,推荐使用 Node.js 流式处理方案,配合前端的打字机效果,用户体验会更加流畅。免费额度完全足够小型客服系统使用,关键是要做好请求频率控制。
正文完
