企业官网建设流程全解析

用 Claude API 构建后端智能问答接口（含完整代码）

搜索"Claude API 接入教程"，你会发现大多数文章讲的是怎么把 Claude Code 这个命令行工具跑起来——但这和"在自己的应用里集成 Claude 问答能力"压根是两回事。本文面向有后端开发基础的工程师，目标是交付一个可以真正跑起来的后端智能问答接口，覆盖 Python（FastAPI）和 Node.js（Express）两套实现，同时把上生产前必须处理的工程问题讲清楚。

一、前置说明：本文适合谁，不适合谁

适合：有 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文件（不要提交到 Git）：

ANTHROPIC_API_KEY=sk-ant-xxxxxxxx

最小验证脚本：

import anthropic import os from dotenv import load_dotenv load_dotenv() client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"]) message = client.messages.create( model="claude-opus-4-5", max_tokens=256, 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 的关键参数和响应结构搞清楚，后面的代码才能读懂。

请求参数

响应体结构

{ "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调大一些。

常见错误码

四、构建后端智能问答接口

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.py）：

import anthropic import os import time client = anthropic.Anthropic(api_key=os.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( model="claude-haiku-4-5", # 问答场景优先用 Haiku 控成本 max_tokens=1024, system=SYSTEM_PROMPT, messages=messages, temperature=0.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.py）：

from 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_code=401, detail="Invalid 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( model="claude-haiku-4-5", max_tokens=1024, system=SYSTEM_PROMPT, messages=messages, ) as stream: for text in stream.text_stream: full_reply.append(text) yield f"data: {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_type="text/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 官网最新定价页为准，价格随时可能调整：

选型上有个实用建议：智能问答场景优先从 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 操作外部系统、或者对接数据库做对话历史持久化。这些方向本文不展开，但基础架构已经打好，后面的路走起来会顺很多。