跳到正文

目录

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

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

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

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

前置条件

  • Python 3.8 及以上
  • 一个 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 初始化

每次 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-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}]
)

timeout=30 是单次请求的秒数上限,max_retries=3 让 SDK 对网络抖动和 429 限流自动重试。这两个参数是生产接入的常见起点,不是越多越好——重试过多会放大下游压力。


发送第一个请求

同步请求

from anthropic import Anthropic
import os

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

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

print(message.content[0].text)

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

参数说明

model

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

模型定位输入输出
claude-haiku-4-5延迟最低,适合实时聊天、简单问答、大批量任务$1$5
claude-sonnet-4-6平衡之选,日常对话、写作、分析的主力模型$3$15
claude-opus-4-6最强能力,适合复杂推理和代码生成$5$25

模型名随版本迭代更新,本文示例以 claude-sonnet-4-6 为准。选模型先看任务对延迟和能力的敏感度:实时交互用 Haiku,兼顾性能与成本用 Sonnet,复杂推理再上 Opus。

max_tokens

控制单次请求最多生成的 token(词元)数。1 token 约等于 0.75 个英文单词或 1-2 个中文字符。按输出长度预期设置:短回答 100-200,几段话 500-1000,完整文章 2000-4096。

message = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=4096,
    messages=[{"role": "user", "content": "写一篇2000字的文章..."}]
)

messages

消息列表,每条消息包含 roleuserassistant)和 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-4-6",
    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-4-6",
    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-4-6"
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 更稳。

停止原因

  • "end_turn":正常完成
  • "max_tokens":达到 max_tokens 限制,响应可能被截断
  • "stop_sequence":遇到指定的停止序列
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 4.6 为例:输入 $3/M,输出 $15/M)
input_cost = (message.usage.input_tokens / 1_000_000) * 3
output_cost = (message.usage.output_tokens / 1_000_000) * 15

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-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Hello"}]
    )
except RateLimitError:
    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-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "我的狗叫豆豆"}]
)
print(response1.content[0].text)

response2 = client.messages.create(
    model="claude-sonnet-4-6",
    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-4-6",
        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=10):
    """只保留最近N轮对话"""
    system_messages = [m for m in messages if m.get("role") == "system"]
    conversation = [m for m in messages if m.get("role") != "system"]

    return system_messages + conversation[-(max_turns * 2):]

messages = trim_conversation(conversation_history, max_turns=5)

摘要旧消息

用 Haiku 模型压缩早期对话,保留关键信息:

def summarize_old_messages(messages, summary_turns=5):
    """将早期对话摘要,保留最近的消息"""
    if len(messages) <= summary_turns * 2 + 2:
        return 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 [
        {"role": "system", "content": f"对话摘要:{summary}"}
    ] + recent

把摘要放进 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-4-6",
    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-4-6",
            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-4-6",
    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-4-6",
    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-4-6",
    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 默认会按指数退避重试。若仍频繁触发,检查是否每次请求都新建了 Anthropic() 实例(应复用同一个 client),以及 max_retries 是否被调小。

400 invalid_request_error:参数不合法

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

响应被截断

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

请求超时

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

模型名与价格过期

文中模型名与定价随版本迭代更新,接入前以 Anthropic 官方模型文档与定价页为准。锁版本时把 anthropic==<版本> 写进依赖,避免升级引入不兼容。


参考资源:

参与讨论

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