用 Claude API 构建后端智能问答接口含完整代码搜索Claude API 接入教程你会发现大多数文章讲的是怎么把 Claude Code 这个命令行工具跑起来——但这和在自己的应用里集成 Claude 问答能力压根是两回事。本文面向有后端开发基础的工程师目标是交付一个可以真正跑起来的后端智能问答接口覆盖 PythonFastAPI和 Node.jsExpress两套实现同时把上生产前必须处理的工程问题讲清楚。一、前置说明本文适合谁不适合谁适合有 Python 或 Node.js 后端开发经验想把 Claude 的问答能力嵌入自己产品、SaaS 或内部工具的开发者。不适合想用 Claude Code CLI 辅助写代码的用户——那是另一个话题本文不涉及。最终交付物一个 HTTP 接口接收用户问题调用 Claude API返回回答支持多轮对话上下文、流式响应、错误重试并对外暴露统一的鉴权机制。二、环境准备与 SDK 安装获取 API Key前往 console.anthropic.com 注册账号在 API Keys 页面创建一个 Key。Key 只在创建时显示一次请立即保存到安全位置。国内网络直连 Anthropic 官方 API 可能存在访问限制。如果需要稳定的国内访问方案可以考虑使用兼容 Anthropic API 格式的第三方接入服务比如 ClaudeAPI 等平台这类平台通常只需替换base_url就能切换主线代码不用改动。具体服务条款、价格和稳定性以各平台官网最新说明为准。Python 方向pip install anthropic fastapi uvicorn python-dotenv redis创建.env文件不要提交到 GitANTHROPIC_API_KEYsk-ant-xxxxxxxx最小验证脚本import anthropic import os from dotenv import load_dotenv load_dotenv() client anthropic.Anthropic(api_keyos.environ[ANTHROPIC_API_KEY]) message client.messages.create( modelclaude-opus-4-5, max_tokens256, messages[{role: user, content: 用一句话解释什么是 REST API}] ) print(message.content[0].text)运行后看到正常回答说明 Key 有效。Node.js 方向npm install anthropic-ai/sdk express dotenv.env文件同上。最小验证脚本import Anthropic from anthropic-ai/sdk; import dotenv/config; const client new Anthropic({ apiKey: process.env.ANTHROPIC_API_KEY }); const message await client.messages.create({ model: claude-opus-4-5, max_tokens: 256, messages: [{ role: user, content: 用一句话解释什么是 REST API }], }); console.log(message.content[0].text);三、核心 API 精讲messages 接口在写接口之前先把messagesAPI 的关键参数和响应结构搞清楚后面的代码才能读懂。请求参数参数类型说明modelstring模型 ID见第七节选型表max_tokensint最大输出 token 数必填建议根据场景设置合理上限systemstring系统提示词定义助手角色和行为边界不参与messages轮次计数messagesarray对话历史role必须交替出现user/assistant且必须以user开头temperaturefloat01越高越随机问答场景建议 0.30.7stop_sequencesarray遇到指定字符串时停止生成可选响应体结构{ id: msg_01XFDUDYJgAACzvnptvVoYEL, type: message, role: assistant, content: [{ type: text, text: 回答内容... }], model: claude-opus-4-5, stop_reason: end_turn, usage: { input_tokens: 42, output_tokens: 128 } }几个关键字段值得注意content[0].text取回答正文usage用于统计费用stop_reason如果是max_tokens说明回答被截断了得考虑把max_tokens调大一些。常见错误码状态码含义处理方式401API Key 无效或过期检查 Key不重试429速率限制指数退避重试见第六节529服务过载同 429 处理500/503服务端错误短暂等待后重试超过阈值报警四、构建后端智能问答接口4.1 Python FastAPI 实现项目结构qa-service/ ├── main.py # FastAPI 应用入口 ├── claude_client.py # Claude 调用封装 ├── session.py # 会话上下文管理 ├── auth.py # 接口鉴权中间件 ├── .env └── requirements.txt会话管理模块session.py先用内存字典做简单实现生产环境再换成 Redis。from collections import defaultdict from typing import List, Dict # 生产环境请替换为 Redis # import redis; r redis.Redis(...) _sessions: Dict[str, List[dict]] defaultdict(list) MAX_HISTORY 20 # 保留最近 20 条防止上下文窗口溢出 def get_history(session_id: str) - List[dict]: return _sessions[session_id] def append_turn(session_id: str, user_msg: str, assistant_msg: str): history _sessions[session_id] history.append({role: user, content: user_msg}) history.append({role: assistant, content: assistant_msg}) # 滑动窗口超出上限时丢弃最早的两条一问一答 if len(history) MAX_HISTORY: _sessions[session_id] history[-MAX_HISTORY:] def clear_session(session_id: str): _sessions.pop(session_id, None)Claude 调用封装claude_client.pyimport anthropic import os import time client anthropic.Anthropic(api_keyos.environ[ANTHROPIC_API_KEY]) SYSTEM_PROMPT 你是一个专业的智能助手回答简洁、准确不确定的内容会明确说明。 def chat(session_id: str, user_message: str, history: list) - tuple[str, dict]: messages history [{role: user, content: user_message}] for attempt in range(3): try: response client.messages.create( modelclaude-haiku-4-5, # 问答场景优先用 Haiku 控成本 max_tokens1024, systemSYSTEM_PROMPT, messagesmessages, temperature0.5, ) text response.content[0].text usage { input_tokens: response.usage.input_tokens, output_tokens: response.usage.output_tokens, } return text, usage except anthropic.RateLimitError: wait 2 ** attempt # 1s, 2s, 4s 指数退避 time.sleep(wait) except anthropic.APIStatusError as e: if e.status_code 500 and attempt 2: time.sleep(1) continue raise raise RuntimeError(Claude API 调用失败已重试 3 次)主路由main.pyfrom fastapi import FastAPI, Header, HTTPException, Depends from pydantic import BaseModel import os from claude_client import chat from session import get_history, append_turn, clear_session app FastAPI() # 简单 API Key 鉴权生产环境建议换成 JWT SERVICE_KEY os.environ.get(SERVICE_API_KEY, change-me) def verify_key(x_api_key: str Header(...)): if x_api_key ! SERVICE_KEY: raise HTTPException(status_code401, detailInvalid API Key) class ChatRequest(BaseModel): session_id: str message: str app.post(/v1/chat, dependencies[Depends(verify_key)]) async def chat_endpoint(req: ChatRequest): history get_history(req.session_id) reply, usage chat(req.session_id, req.message, history) append_turn(req.session_id, req.message, reply) return {reply: reply, usage: usage} app.delete(/v1/session/{session_id}, dependencies[Depends(verify_key)]) async def clear(session_id: str): clear_session(session_id) return {status: cleared}启动命令uvicorn main:app --host 0.0.0.0 --port 8000测试curl -X POST http://localhost:8000/v1/chat \ -H Content-Type: application/json \ -H x-api-key: change-me \ -d {session_id: user_001, message: 什么是向量数据库}4.2 Node.js Express 实现核心逻辑和 Python 版等价适合前端同学或已有 Node.js 技术栈的团队// app.js import express from express; import Anthropic from anthropic-ai/sdk; import dotenv/config; const app express(); app.use(express.json()); const client new Anthropic({ apiKey: process.env.ANTHROPIC_API_KEY }); const sessions new Map(); // 生产环境替换为 Redis function authMiddleware(req, res, next) { if (req.headers[x-api-key] ! process.env.SERVICE_API_KEY) { return res.status(401).json({ error: Invalid API Key }); } next(); } app.post(/v1/chat, authMiddleware, async (req, res) { const { session_id, message } req.body; const history sessions.get(session_id) || []; const response await client.messages.create({ model: claude-haiku-4-5, max_tokens: 1024, system: 你是一个专业的智能助手回答简洁、准确。, messages: [...history, { role: user, content: message }], }); const reply response.content[0].text; const newHistory [ ...history, { role: user, content: message }, { role: assistant, content: reply }, ].slice(-20); // 滑动窗口 sessions.set(session_id, newHistory); res.json({ reply, usage: response.usage }); }); app.listen(8000, () console.log(Running on :8000));五、流式响应实现打字机效果非流式接口要等模型生成完才一次性返回首字延迟很高。流式响应通过 SSE 逐 token 推送用户体验明显更好。Python 流式路由from fastapi.responses import StreamingResponse import json app.post(/v1/chat/stream, dependencies[Depends(verify_key)]) async def chat_stream(req: ChatRequest): history get_history(req.session_id) messages history [{role: user, content: req.message}] def generate(): full_reply [] with client.messages.stream( modelclaude-haiku-4-5, max_tokens1024, systemSYSTEM_PROMPT, messagesmessages, ) as stream: for text in stream.text_stream: full_reply.append(text) yield fdata: {json.dumps({delta: text})}\n\n # 流结束后保存上下文 append_turn(req.session_id, req.message, .join(full_reply)) yield data: [DONE]\n\n return StreamingResponse(generate(), media_typetext/event-stream)前端接收约 15 行const es new EventSource(/v1/chat/stream); // 实际需携带鉴权可用 fetch ReadableStream // 推荐用 fetch 替代 EventSource 以支持自定义 Header const resp await fetch(/v1/chat/stream, { method: POST, headers: { Content-Type: application/json, x-api-key: ... }, body: JSON.stringify({ session_id: user_001, message: 你好 }), }); const reader resp.body.getReader(); const decoder new TextDecoder(); while (true) { const { done, value } await reader.read(); if (done) break; const chunk decoder.decode(value); // 解析 SSE data 字段并追加到 UI console.log(chunk); }六、生产级注意事项API Key 安全Key 只存在于后端环境变量永远不出现在前端代码、Git 仓库或日志里。前端只调用你自建的接口由后端持有并使用 Key这一点没有商量余地。上下文窗口管理上面代码里的滑动窗口保留最近 20 条是最简单的方案。如果对话连贯性要求比较高可以在截断前先调用 Claude 对历史做一次摘要压缩用摘要替代早期记录效果会好不少。速率限制重试第四节的claude_client.py已经包含了指数退避实现1s → 2s → 4s。生产环境建议加监控告警连续重试失败时及时通知运维。token 用量监控每次调用的usage.input_tokens和usage.output_tokens写入数据库按用户或会话聚合便于费用分析和异常用量告警。这个习惯养成早一点后面排查问题会省很多力气。七、模型选型与成本参考以下是各模型能力定位参考具体价格以 Anthropic 官网最新定价页为准价格随时可能调整模型定位适用场景claude-haiku-4-5速度最快、成本最低高频简单问答、分类、摘要claude-sonnet-4-5能力与成本均衡复杂推理、代码生成、多步任务claude-opus-4-5能力最强、成本最高高价值低频任务、需要深度分析选型上有个实用建议智能问答场景优先从 Haiku 开始上线后通过 A/B 测试评估回答质量质量不达标再升级到 Sonnet。Opus 一般只用于离线批处理或对质量要求极高的专业场景日常问答用它有点大材小用。八、完整运行说明Python 版本完整启动流程git clone your-repo cd qa-service pip install -r requirements.txt # 配置环境变量 cp .env.example .env # 编辑 .env填入 ANTHROPIC_API_KEY 和 SERVICE_API_KEY uvicorn main:app --reload接口联调测试# 第一轮问答 curl -X POST http://localhost:8000/v1/chat \ -H Content-Type: application/json \ -H x-api-key: your-service-key \ -d {session_id: test_001, message: 什么是 Token} # 第二轮上下文延续 curl -X POST http://localhost:8000/v1/chat \ -H Content-Type: application/json \ -H x-api-key: your-service-key \ -d {session_id: test_001, message: 它和字符有什么区别} # 清除会话 curl -X DELETE http://localhost:8000/v1/session/test_001 \ -H x-api-key: your-service-key第二轮请求不需要重复传问题背景服务端会自动携带上下文——这其实是智能问答接口和单次调用最核心的区别所在。至此一个具备多轮上下文、流式响应、接口鉴权和基础错误处理的后端智能问答接口已经完整呈现。后续可以在此基础上继续扩展接入知识库做 RAG 检索、加入函数调用Tool Use让 Claude 操作外部系统、或者对接数据库做对话历史持久化。这些方向本文不展开但基础架构已经打好后面的路走起来会顺很多。