跳到正文

目录

Claude API 基础专题(一):认证、请求与会话管理

Claude API 基础专题(一):认证、请求与会话管理

Claude Messages API(应用程序接口)的入口是一个 messages.create() 调用:给它模型名、消息列表和最大输出长度,它返回一条完整的回复。本文把这套调用的工程细节拆开讲——密钥怎么管、请求怎么发、响应怎么解析、多轮对话怎么维护、系统提示词怎么写、结构化输出怎么拿到合法 JSON。代码基于 anthropic Python SDK(软件开发包),示例模型统一用 claude-sonnet-5(2026 年 9 月口径,模型迭代快,接入前以官方模型页为准)。

读完本文,你能拿到一份可以直接改着用的请求模板,以及排查 401、429、输出截断这类常见问题时的判断顺序。只关心某个环节时,按标题跳读即可。

前置条件

  • Python 3.10 及以上(anthropic SDK 的硬性要求,见其 pyproject.toml 的 requires-python)
  • 一个 Anthropic Console 账户和 API 密钥
  • 已安装 anthropic SDK:
pip install anthropic

认证与密钥管理

获取 API 密钥

  1. 访问 Anthropic Console
  2. 注册账户
  3. 在「API Keys」页面创建新密钥
  4. 复制密钥并妥善保存

密钥管理

密钥只存在于运行它的环境里,不写进代码。一旦提交到 Git 仓库,即使后续删除,历史记录中仍可追溯。

开发环境:从环境变量读取

import os
from anthropic import Anthropic

api_key = os.environ.get("ANTHROPIC_API_KEY")
if not api_key:
    raise ValueError("ANTHROPIC_API_KEY 环境变量未设置")

client = Anthropic(api_key=api_key)

开发环境推荐:.env 文件

# .env 文件(不要提交到 Git!)
ANTHROPIC_API_KEY=<your-real-key>
from dotenv import load_dotenv
import os

load_dotenv()
api_key = os.environ.get("ANTHROPIC_API_KEY")

from anthropic import Anthropic
client = Anthropic(api_key=api_key)
pip install python-dotenv

生产环境:云密钥管理服务

import boto3
import json
from anthropic import Anthropic

secret_name = "anthropic-api-key"
region_name = "us-east-1"

session = boto3.session.Session()
client_secrets = session.client(
    service_name='secretsmanager',
    region_name=region_name
)

response = client_secrets.get_secret_value(SecretId=secret_name)
api_key = json.loads(response['SecretString'])['api_key']

anthropic_client = Anthropic(api_key=api_key)

SDK 初始化

不传 api_key 时,SDK 会自动读取 ANTHROPIC_API_KEY 环境变量,所以最简写法就是 client = Anthropic()。前面几节显式传参是为了让"密钥从哪来"一目了然。

每次 Anthropic() 都会建立新的连接池。同一个进程里复用一个客户端实例,避免反复建连:

from anthropic import Anthropic
import os

class AnthropicClient:
    """Anthropic API 客户端封装"""

    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._client = Anthropic(
                api_key=os.environ.get("ANTHROPIC_API_KEY"),
                timeout=30,
                max_retries=3,
            )
        return cls._instance

    @property
    def client(self):
        return self._client

# 使用单例模式
anthropic = AnthropicClient()
response = anthropic.client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}]
)

timeout=30 是单次请求的秒数上限(SDK 默认 10 分钟),max_retries=3 让 SDK 对网络抖动和 429 限流自动重试(SDK 默认 2 次,退避间隔从 0.5 秒起步、上限 8 秒)。这两个参数是生产接入的常见起点,不是越多越好——重试过多会放大下游压力。


发送第一个请求

同步请求

from anthropic import Anthropic
import os

client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

message = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": "用一句话解释什么是量子计算"
        }
    ]
)

print(message.content[0].text)

message.content 是内容块列表,取 [0].text 就是模型生成的文本。运行这段代码,终端打印出回复,就说明密钥、SDK、网络链路都通了。

参数说明

model

在售模型按能力和成本递增排序(价格为每百万 token(词元),输入/输出,2026 年 9 月口径,随版本调整,接入前以官方定价页为准):

模型定位输入输出
claude-haiku-4-5延迟最低,适合实时聊天、简单问答、大批量任务$1$5
claude-sonnet-5平衡之选,日常对话、写作、分析的主力模型$2$10
claude-opus-5官方建议的大多数工作负载起点,复杂推理和代码生成$5$25
claude-fable-5-1高阶推理与长程智能体任务$10$50

Sonnet 4.6、Opus 4.6 等旧模型已转入 legacy(历史型号),API 仍可调用,价格官网可查。模型名随版本迭代更新,本文示例统一用 claude-sonnet-5。选模型先看任务对延迟和能力的敏感度:实时交互用 Haiku,兼顾性能与成本用 Sonnet,复杂推理用 Opus,长程智能体任务再上 Fable。

max_tokens

控制单次请求最多生成的 token 数。粗略换算:1 token 约合 0.75 个英文单词,中文约 1-2 个字——这是旧模型的口径,Opus 4.7 起换用新 tokenizer,同样文本比旧模型多算约 30%,跨模型估算成本时别直接搬数字,用 client.messages.count_tokens() 按目标模型实测。按输出长度预期设置 max_tokens:短回答 100-200,几段话 500-1000,完整文章 2000-4096。

message = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=4096,
    messages=[{"role": "user", "content": "写一篇2000字的文章介绍量子计算"}]
)

messages

消息列表,每条消息包含 role(user 或 assistant)和 content:

messages=[
    {"role": "user", "content": "什么是Python?"},
    {"role": "assistant", "content": "Python是一种高级编程语言..."},
    {"role": "user", "content": "它适合做什么?"}
]

流式响应

from anthropic import Anthropic
import os

client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "讲一个关于程序员的笑话"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    print()

长文本生成时,非流式模式用户需等待数秒到十几秒。流式响应是生产场景的推荐做法——首字更快到达,用户不用干等整段生成完。


理解响应结构

Message 对象

message = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "解释光合作用"}]
)

print(message.id)          # msg_xxxxx
print(message.type)        # "message"
print(message.role)        # "assistant"
print(message.content)     # [ContentBlock(text='...')]
print(message.model)       # "claude-sonnet-5"
print(message.stop_reason) # "end_turn"
print(message.stop_sequence) # None
print(message.usage)       # Usage(input_tokens=xx, output_tokens=xx)

解析内容

for block in message.content:
    if block.type == "text":
        print(block.text)

content 不一定是纯文本:模型可能返回工具调用块或思考块。逐块判断 type,比直接取 [0].text 更稳。

停止原因

stop_reason 共五种取值:

  • "end_turn":模型自然讲完,正常完成
  • "max_tokens":达到 max_tokens 限制,响应可能被截断
  • "stop_sequence":遇到请求里指定的停止序列
  • "pause_turn":长回合被暂停,把响应原样放进下一轮请求可让模型继续
  • "refusal":模型因安全策略拒绝回答,输出不保证符合你要求的格式
if message.stop_reason == "max_tokens":
    print("响应被截断,建议增加max_tokens值")
elif message.stop_reason == "end_turn":
    print("响应正常完成")

Token 使用量

usage 给出本次请求消耗的输入和输出 token,是计算成本、优化提示词长度的依据。

print(f"输入token: {message.usage.input_tokens}")
print(f"输出token: {message.usage.output_tokens}")
print(f"总token: {message.usage.input_tokens + message.usage.output_tokens}")

# 计算成本(以 Sonnet 5 为例:输入 $2/M,输出 $10/M)
input_cost = (message.usage.input_tokens / 1_000_000) * 2
output_cost = (message.usage.output_tokens / 1_000_000) * 10

print(f"本次请求成本: ${input_cost + output_cost:.6f}")

错误处理

from anthropic import Anthropic, RateLimitError, APIError
import os

client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

try:
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Hello"}]
    )
except RateLimitError:
    # SDK 已自动退避重试过,走到这里说明重试次数内始终 429
    print("速率限制:请求太频繁,等待后重试")
    import time
    time.sleep(5)
except APIError as e:
    print(f"API 错误: {e}")
except Exception as e:
    print(f"未知错误: {e}")

按异常类型分分支处理,别把限流和参数错误混在一起:限流适合退避重试,参数错误重试多少次都是同样的结果。兜底的 Exception 分支只用于记录日志,不要在这里吞掉错误继续业务。


多轮对话与会话管理

Claude API 本身是无状态的——每次 messages.create() 调用都是独立的。会话由客户端维护的消息列表定义。

无状态(无记忆):

response1 = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "我的狗叫豆豆"}]
)
print(response1.content[0].text)

response2 = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "它喜欢吃什么?"}]
)
# Claude 不记得“豆豆”

有状态(有记忆):

conversation_history = []

while True:
    user_input = input("你: ")

    conversation_history.append({"role": "user", "content": user_input})

    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=1024,
        messages=conversation_history
    )

    assistant_message = response.content[0].text
    conversation_history.append({"role": "assistant", "content": assistant_message})

    print(f"Claude: {assistant_message}")

关键在最后一步:把模型这次的回复追加回 conversation_history,下一轮它才看得到自己说过什么。

会话管理技巧

对话越长,token 消耗越大,模型也越容易抓不住重点。下面三种做法按成本从低到高。

限制历史长度

def trim_conversation(messages, max_turns=5):
    """只保留最近 N 轮对话(一轮 = 一条 user 加一条 assistant)"""
    return messages[-(max_turns * 2):]

messages = trim_conversation(conversation_history, max_turns=5)

摘要旧消息

用 Haiku 模型压缩早期对话,保留关键信息。注意摘要不能以 system 角色塞回 messages——Messages API 的输入消息只有 user 和 assistant 两种角色,系统级内容必须走顶层 system 参数:

def summarize_old_messages(messages, summary_turns=5):
    """摘要早期对话,返回 (摘要文本, 最近消息列表)"""
    if len(messages) <= summary_turns * 2 + 2:
        return None, messages

    early = messages[:-summary_turns * 2]
    recent = messages[-summary_turns * 2:]

    early_text = "\n".join([f"{m['role']}: {m['content']}" for m in early])

    summary_prompt = f"""将以下对话摘要成一段话,保留关键信息:

{early_text}

摘要:"""

    summary_response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=500,
        messages=[{"role": "user", "content": summary_prompt}]
    )

    summary = summary_response.content[0].text
    return summary, recent

summary, recent = summarize_old_messages(conversation_history, summary_turns=5)
kwargs = {"system": f"对话摘要:{summary}"} if summary else {}
response = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=recent,
    **kwargs
)

把摘要放进 system 参数,既让它在整轮对话里持续生效,又不占用最近几轮消息的位置。

分离话题

class ConversationManager:
    """会话管理器:支持多话题"""

    def __init__(self):
        self.conversations = {}
        self.current_id = None

    def start_new(self, conversation_id):
        self.current_id = conversation_id
        self.conversations[conversation_id] = []

    def add_message(self, role, content):
        if self.current_id is None:
            self.start_new("default")
        self.conversations[self.current_id].append({
            "role": role,
            "content": content
        })

    def get_messages(self, conversation_id=None):
        cid = conversation_id or self.current_id
        return self.conversations.get(cid, [])

    def switch_conversation(self, conversation_id):
        if conversation_id not in self.conversations:
            self.conversations[conversation_id] = []
        self.current_id = conversation_id

# 使用示例
manager = ConversationManager()
manager.start_new("技术支持")
manager.add_message("user", "我的代码报错了")
manager.add_message("assistant", "请告诉我错误信息")
manager.add_message("user", "NameError: name 'x' is not defined")

manager.start_new("产品咨询")
manager.add_message("user", "你们的产品有什么特点")

manager.switch_conversation("技术支持")
messages = manager.get_messages()

多话题场景把每个会话的上下文分桶管理,互不串扰。


系统提示词

系统提示词(System Prompt)设置 AI 的行为和角色,作用于整个对话。

response = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1024,
    system="你是一位专业的产品经理,用词简洁专业。",
    messages=[{"role": "user", "content": "我应该做什么产品?"}]
)

它和 messages 是独立的参数,不进历史列表,也不占用对话轮次。

常见模式

角色设定

system = """你是一位拥有20年经验的高级Python工程师。
你的特点:
- 代码风格遵循PEP 8
- 喜欢用类型提示
- 注重性能优化
- 说话直接,有话直说
"""

输出格式指定

system = """你是一个数据分析师。

回答问题时必须使用以下格式:

## 主要发现
[最重要的1-2个发现]

## 详细分析
[详细的分析内容]

## 建议
[基于分析的可执行建议]

## 数据来源
[使用的数据]
"""

约束条件

system = """你是一位财经记者。

约束条件:
- 不预测具体股价
- 引用数据时注明来源
- 风险提示必须清晰
- 不使用"一定"、"保证"等绝对词汇
"""

示例注入(Few-shot in system)

system = """你是一个翻译助手。

翻译示例:
- "Hello, how are you?" → "你好,你怎么样?"
- "The weather is nice today." → "今天天气很好。"

注意:
- 中文翻译用"你"而不是"您"
- 保持原文的语气和情感
"""

设计要点

系统提示词应具体、一致、可验证。避免相互矛盾的指令(如同时要求"诚实"和"必要时可以说善意的谎言"),以及过于模糊的设定(如"你是 AI 助手,回答用户问题")。角色、格式、约束、示例四类模式可以组合,但每一类都要能直接对照检查输出是否符合。

测试系统提示词

下面这个函数把一批测试输入跑一遍,逐条打印输出,适合在调整提示词时做回归对比。假设 client 已在上面定义:

def test_system_prompt(system_prompt, test_cases):
    """测试系统提示词"""
    for i, test in enumerate(test_cases):
        response = client.messages.create(
            model="claude-sonnet-5",
            max_tokens=500,
            system=system_prompt,
            messages=[{"role": "user", "content": test}]
        )
        print(f"测试{i+1}: {test}")
        print(f"响应: {response.content[0].text[:200]}...")
        print("-" * 50)

结构化输出

需要 AI 返回 JSON 等特定格式数据时,有几种方案,可靠性从低到高。只有结构化输出(方法 2、3)能保证输出格式合法,其余方案都要靠自己的代码兜底。

方法 1:提示词中要求 JSON

在提示词里描述期望的 JSON 结构,再手动解析返回文本。实现最直接,但 Claude 可能包裹代码块、加多余文本、漏字段或类型不对,需要自己清洗和重试。

response = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1000,
    messages=[{
        "role": "user",
        "content": """返回一个JSON对象,包含水果信息:
        {"name": "水果名", "color": "颜色", "taste": "味道"}"""
    }]
)

import json
text = response.content[0].text
if "```json" in text:
    text = text.split("```json")[1].split("```")[0]
elif "```" in text:
    text = text.split("```")[1].split("```")[0]

data = json.loads(text.strip())
print(data)

方法 2:结构化输出(output_config.format)

用 output_config.format 声明 JSON Schema(模式),Claude 通过受限解码保证输出是合法 JSON 且字段类型、必填项符合 schema,不再出现 json.loads 报错或字段缺失。

response = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1000,
    messages=[{
        "role": "user",
        "content": "返回3个编程语言的列表"
    }],
    output_config={
        "format": {
            "type": "json_schema",
            "schema": {
                "type": "object",
                "properties": {
                    "languages": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "name": {"type": "string"},
                                "year": {"type": "integer"},
                                "paradigm": {"type": "string"}
                            }
                        }
                    }
                },
                "required": ["languages"],
                "additionalProperties": False
            }
        }
    }
)

import json
data = json.loads(response.content[0].text)
print(data)

几个要点:

  • 返回的 JSON 齐全时,直接把 response.content[0].text 交给 json.loads 即可,无需再清洗。
  • 结构化输出保证的是格式合规,不保证内容正确。字段类型、必填项一定符合 schema,但值是否合理、事实是否准确仍要自己判断。
  • 首次使用某个 schema 会有一次额外的语法编译延迟,之后会缓存约 24 小时,第二次起明显变快。
  • schema 里 required 的字段会排在 optional 之前输出,若字段顺序对下游重要,把字段都设为必填或在解析时按关键词取值。

方法 3:Pydantic + messages.parse()

不写原始 JSON Schema,用 Pydantic 模型声明结构,配合 SDK 的 client.messages.parse(),返回的 response.parsed_output 直接是校验过的模型实例。

from pydantic import BaseModel
from anthropic import Anthropic
import os

class Language(BaseModel):
    name: str
    year: int
    paradigm: str

class LanguageList(BaseModel):
    languages: list[Language]

client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

response = client.messages.parse(
    model="claude-sonnet-5",
    max_tokens=1000,
    messages=[{"role": "user", "content": "返回3个编程语言的列表"}],
    output_format=LanguageList,
)

print(response.parsed_output)
print(response.parsed_output.languages[0].name)

output_format 接受一个 Pydantic 模型类,SDK 会把它转成 JSON Schema 传给 output_config.format,再用同一个模型校验返回结果。类型错误在解析阶段就会被拦截,省去手写 schema 和手动验证。

边界情况

结构化输出在两种情况下可能不满足 schema:模型因安全原因拒绝回答(stop_reason 为 "refusal"),或输出被 max_tokens 截断(stop_reason 为 "max_tokens")。前者按拒绝处理,后者调大 max_tokens 重试。

if message.stop_reason == "refusal":
    print("模型拒绝回答")
elif message.stop_reason == "max_tokens":
    print("输出被截断,建议增加max_tokens")

如果仍需手动解析不可靠的 JSON 文本(比如方法 1 的产物),可以用一个容错解析函数兜底:

def safe_json_parse(text):
    """安全解析JSON,处理代码块和多余文本"""
    import json
    import re

    text = re.sub(r'```json\s*', '', text)
    text = re.sub(r'```\s*$', '', text)
    text = text.strip()

    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass

    start = text.find('{')
    end = text.rfind('}') + 1
    if start != -1 and end > start:
        try:
            return json.loads(text[start:end])
        except json.JSONDecodeError:
            pass

    return None

常见问题与排查

401 authentication_error:密钥无效

请求返回 authentication_error 时,先确认 ANTHROPIC_API_KEY 真的被读到了——本地 .env 文件没被加载是最常见的原因。其次检查密钥是否过期或被轮换,以及是否写错成了环境变量名。

429 rate_limit_error:请求过频

SDK 默认按指数退避自动重试(最多 2 次)。若仍频繁触发,检查是否每次请求都新建了 Anthropic() 实例(应复用同一个 client),以及 max_retries 是否被调小;持续 429 说明撞到了账户或模型的速率上限,去 Console 核对限额或申请提额。

400 invalid_request_error:参数不合法

通常是 messages 结构不对或 model 名过期。先对照错误信息里的字段名定位,再核对 Messages API 文档。

响应被截断

回复中途断掉且 stop_reason 是 max_tokens,说明 max_tokens 设小了。调大后重试;输出本身很长时,改用流式读取,用户不用等整段生成完。

请求超时

短请求频繁超时,先检查网络代理或防火墙,再考虑调小 timeout 并配合重试。注意 timeout 与 max_retries 是乘数关系:重试会放大总耗时。

模型名与价格过期

文中模型名与定价随版本迭代更新(新模型上线,旧型号逐步转入 legacy,API ID 仍可调用),接入前以 Anthropic 官方模型概览与定价页为准。锁版本时把 anthropic==<版本> 写进依赖,避免升级引入不兼容。


参考资源:

参与讨论

使用 GitHub 登录。欢迎补充事实、异议与实践。