Back to Details

anthropic-sdk

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Anthropic SDK - Official Claude AI Integration

Anthropic SDK - Claude AI官方集成

progressive_disclosure: entry_point: summary: "Official Anthropic SDK for Claude AI - chat, streaming, function calling, vision" when_to_use: - "When integrating Claude AI into applications" - "When building AI-powered features with Claude models" - "When using function calling/tool use patterns" - "When processing images with vision models" - "When implementing streaming chat interfaces" quick_start: - "pip install anthropic (Python) or npm install @anthropic-ai/sdk (TypeScript)" - "Set ANTHROPIC_API_KEY environment variable" - "Create client and send messages with Messages API" - "Use streaming for real-time responses" installation: python: "pip install anthropic" typescript: "npm install @anthropic-ai/sdk" config: - "ANTHROPIC_API_KEY: Your API key from console.anthropic.com" - "Model: claude-3-5-sonnet-20241022 (recommended)" - "Max tokens: 1024-8192 for responses" token_estimate: entry: 85 full: 5000

progressive_disclosure: entry_point: summary: "Anthropic官方Claude AI SDK - 支持聊天、流式传输、函数调用、视觉功能" when_to_use: - "将Claude AI集成到应用程序中时" - "基于Claude模型构建AI驱动的功能时" - "使用函数调用/工具调用模式时" - "使用视觉模型处理图片时" - "实现流式聊天界面时" quick_start: - "pip install anthropic（Python）或npm install @anthropic-ai/sdk（TypeScript）" - "设置ANTHROPIC_API_KEY环境变量" - "创建客户端并通过Messages API发送消息" - "使用流式传输获取实时响应" installation: python: "pip install anthropic" typescript: "npm install @anthropic-ai/sdk" config: - "ANTHROPIC_API_KEY：从console.anthropic.com获取的API密钥" - "Model：claude-3-5-sonnet-20241022（推荐）" - "Max tokens：响应长度设置为1024-8192" token_estimate: entry: 85 full: 5000

Installation & Setup

安装与设置

Python

Python

bash
pip install anthropic
bash
pip install anthropic

TypeScript

TypeScript

bash
npm install @anthropic-ai/sdk
bash
npm install @anthropic-ai/sdk

API Key Configuration

API密钥配置

bash
export ANTHROPIC_API_KEY='your-api-key-here'
Get your API key from: https://console.anthropic.com/settings/keys
bash
export ANTHROPIC_API_KEY='your-api-key-here'
获取API密钥：https://console.anthropic.com/settings/keys

Messages API - Basic Usage

Messages API - 基础用法

Python - Simple Message

Python - 简单消息

python
import anthropic
import os

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

message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Explain quantum computing in simple terms"}
    ]
)

print(message.content[0].text)
python
import anthropic
import os

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

message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "用简单的语言解释量子计算"}
    ]
)

print(message.content[0].text)

TypeScript - Simple Message

TypeScript - 简单消息

typescript
import Anthropic from '@anthropic-ai/sdk';

const client = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY,
});

const message = await client.messages.create({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  messages: [
    { role: 'user', content: 'Explain quantum computing in simple terms' }
  ],
});

console.log(message.content[0].text);
typescript
import Anthropic from '@anthropic-ai/sdk';

const client = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY,
});

const message = await client.messages.create({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  messages: [
    { role: 'user', content: '用简单的语言解释量子计算' }
  ],
});

console.log(message.content[0].text);

System Prompts

系统提示词

python
undefined
python
undefined

Python - System prompt for context

Python - 带上下文的系统提示词

message = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=1024, system="You are a helpful coding assistant specializing in Python and TypeScript.", messages=[ {"role": "user", "content": "How do I handle errors in async functions?"} ] )

```typescript
// TypeScript - System prompt
const message = await client.messages.create({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  system: 'You are a helpful coding assistant specializing in Python and TypeScript.',
  messages: [
    { role: 'user', content: 'How do I handle errors in async functions?' }
  ],
});
message = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=1024, system="你是一位精通Python和TypeScript的编程助手。", messages=[ {"role": "user", "content": "如何处理异步函数中的错误？"} ] )

```typescript
// TypeScript - 系统提示词
const message = await client.messages.create({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  system: '你是一位精通Python和TypeScript的编程助手。',
  messages: [
    { role: 'user', content: '如何处理异步函数中的错误？' }
  ],
});

Streaming Responses

流式响应

Python - Streaming

Python - 流式传输

python
undefined
python
undefined

Real-time streaming responses

实时流式响应

with client.messages.stream( model="claude-3-5-sonnet-20241022", max_tokens=1024, messages=[ {"role": "user", "content": "Write a short poem about coding"} ] ) as stream: for text in stream.text_stream: print(text, end="", flush=True)
undefined
with client.messages.stream( model="claude-3-5-sonnet-20241022", max_tokens=1024, messages=[ {"role": "user", "content": "写一首关于编程的短诗"} ] ) as stream: for text in stream.text_stream: print(text, end="", flush=True)
undefined

Python - Async Streaming

Python - 异步流式传输

python
import asyncio

async def stream_response():
    async with client.messages.stream(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": "Explain recursion"}
        ]
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)

asyncio.run(stream_response())
python
import asyncio

async def stream_response():
    async with client.messages.stream(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": "解释递归"}
        ]
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)

asyncio.run(stream_response())

TypeScript - Streaming

TypeScript - 流式传输

typescript
// Streaming with event handlers
const stream = await client.messages.stream({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  messages: [
    { role: 'user', content: 'Write a short poem about coding' }
  ],
});

for await (const chunk of stream) {
  if (chunk.type === 'content_block_delta' &&
      chunk.delta.type === 'text_delta') {
    process.stdout.write(chunk.delta.text);
  }
}
typescript
// 使用事件处理器的流式传输
const stream = await client.messages.stream({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  messages: [
    { role: 'user', content: '写一首关于编程的短诗' }
  ],
});

for await (const chunk of stream) {
  if (chunk.type === 'content_block_delta' &&
      chunk.delta.type === 'text_delta') {
    process.stdout.write(chunk.delta.text);
  }
}

Function Calling / Tool Use

函数调用/工具使用

Python - Function Calling

Python - 函数调用

python
undefined
python
undefined

Define tools (functions)

定义工具（函数）

tools = [ { "name": "get_weather", "description": "Get the current weather for a location", "input_schema": { "type": "object", "properties": { "location": { "type": "string", "description": "City name, e.g., San Francisco, CA" }, "unit": { "type": "string", "enum": ["celsius", "fahrenheit"], "description": "Temperature unit" } }, "required": ["location"] } } ]
tools = [ { "name": "get_weather", "description": "获取指定地点的当前天气", "input_schema": { "type": "object", "properties": { "location": { "type": "string", "description": "城市名称，例如：San Francisco, CA" }, "unit": { "type": "string", "enum": ["celsius", "fahrenheit"], "description": "温度单位" } }, "required": ["location"] } } ]

Initial request

初始请求

message = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=1024, tools=tools, messages=[ {"role": "user", "content": "What's the weather in San Francisco?"} ] )
message = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=1024, tools=tools, messages=[ {"role": "user", "content": "旧金山的天气怎么样？"} ] )

Check for tool use

检查是否需要使用工具

if message.stop_reason == "tool_use": tool_use = next(block for block in message.content if block.type == "tool_use") tool_name = tool_use.name tool_input = tool_use.input
# Execute function (mock example)
if tool_name == "get_weather":
    weather_result = {
        "temperature": 72,
        "unit": "fahrenheit",
        "conditions": "sunny"
    }

# Send result back to Claude
response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    tools=tools,
    messages=[
        {"role": "user", "content": "What's the weather in San Francisco?"},
        {"role": "assistant", "content": message.content},
        {
            "role": "user",
            "content": [
                {
                    "type": "tool_result",
                    "tool_use_id": tool_use.id,
                    "content": str(weather_result)
                }
            ]
        }
    ]
)
print(response.content[0].text)
undefined
if message.stop_reason == "tool_use": tool_use = next(block for block in message.content if block.type == "tool_use") tool_name = tool_use.name tool_input = tool_use.input
# 执行函数（模拟示例）
if tool_name == "get_weather":
    weather_result = {
        "temperature": 72,
        "unit": "fahrenheit",
        "conditions": "sunny"
    }

# 将结果返回给Claude
response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    tools=tools,
    messages=[
        {"role": "user", "content": "旧金山的天气怎么样？"},
        {"role": "assistant", "content": message.content},
        {
            "role": "user",
            "content": [
                {
                    "type": "tool_result",
                    "tool_use_id": tool_use.id,
                    "content": str(weather_result)
                }
            ]
        }
    ]
)
print(response.content[0].text)
undefined

TypeScript - Function Calling

TypeScript - 函数调用

typescript
// Define tools
const tools: Anthropic.Tool[] = [
  {
    name: 'get_weather',
    description: 'Get the current weather for a location',
    input_schema: {
      type: 'object',
      properties: {
        location: {
          type: 'string',
          description: 'City name, e.g., San Francisco, CA',
        },
        unit: {
          type: 'string',
          enum: ['celsius', 'fahrenheit'],
          description: 'Temperature unit',
        },
      },
      required: ['location'],
    },
  },
];

// Initial request
const message = await client.messages.create({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  tools,
  messages: [
    { role: 'user', content: "What's the weather in San Francisco?" },
  ],
});

// Check for tool use
if (message.stop_reason === 'tool_use') {
  const toolUse = message.content.find(
    (block): block is Anthropic.ToolUseBlock => block.type === 'tool_use'
  );

  if (toolUse && toolUse.name === 'get_weather') {
    // Execute function
    const weatherResult = {
      temperature: 72,
      unit: 'fahrenheit',
      conditions: 'sunny',
    };

    // Send result back
    const response = await client.messages.create({
      model: 'claude-3-5-sonnet-20241022',
      max_tokens: 1024,
      tools,
      messages: [
        { role: 'user', content: "What's the weather in San Francisco?" },
        { role: 'assistant', content: message.content },
        {
          role: 'user',
          content: [
            {
              type: 'tool_result',
              tool_use_id: toolUse.id,
              content: JSON.stringify(weatherResult),
            },
          ],
        },
      ],
    });

    console.log(response.content[0].text);
  }
}
typescript
// 定义工具
const tools: Anthropic.Tool[] = [
  {
    name: 'get_weather',
    description: '获取指定地点的当前天气',
    input_schema: {
      type: 'object',
      properties: {
        location: {
          type: 'string',
          description: '城市名称，例如：San Francisco, CA',
        },
        unit: {
          type: 'string',
          enum: ['celsius', 'fahrenheit'],
          description: '温度单位',
        },
      },
      required: ['location'],
    },
  },
];

// 初始请求
const message = await client.messages.create({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  tools,
  messages: [
    { role: 'user', content: "旧金山的天气怎么样？" },
  ],
});

// 检查是否需要使用工具
if (message.stop_reason === 'tool_use') {
  const toolUse = message.content.find(
    (block): block is Anthropic.ToolUseBlock => block.type === 'tool_use'
  );

  if (toolUse && toolUse.name === 'get_weather') {
    // 执行函数
    const weatherResult = {
      temperature: 72,
      unit: 'fahrenheit',
      conditions: 'sunny',
    };

    // 将结果返回
    const response = await client.messages.create({
      model: 'claude-3-5-sonnet-20241022',
      max_tokens: 1024,
      tools,
      messages: [
        { role: 'user', content: "旧金山的天气怎么样？" },
        { role: 'assistant', content: message.content },
        {
          role: 'user',
          content: [
            {
              type: 'tool_result',
              tool_use_id: toolUse.id,
              content: JSON.stringify(weatherResult),
            },
          ],
        },
      ],
    });

    console.log(response.content[0].text);
  }
}

Vision Models - Image Input

视觉模型 - 图片输入

Python - Image Analysis

Python - 图片分析

python
import base64
python
import base64

Load image

加载图片

with open("image.jpg", "rb") as image_file: image_data = base64.standard_b64encode(image_file.read()).decode("utf-8")
with open("image.jpg", "rb") as image_file: image_data = base64.standard_b64encode(image_file.read()).decode("utf-8")

Send image to Claude

将图片发送给Claude

message = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=1024, messages=[ { "role": "user", "content": [ { "type": "image", "source": { "type": "base64", "media_type": "image/jpeg", "data": image_data, }, }, { "type": "text", "text": "Describe this image in detail" } ], } ], )
print(message.content[0].text)
undefined
message = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=1024, messages=[ { "role": "user", "content": [ { "type": "image", "source": { "type": "base64", "media_type": "image/jpeg", "data": image_data, }, }, { "type": "text", "text": "详细描述这张图片" } ], } ], )
print(message.content[0].text)
undefined

TypeScript - Image Analysis

TypeScript - 图片分析

typescript
import * as fs from 'fs';

// Load image
const imageData = fs.readFileSync('image.jpg').toString('base64');

// Send image to Claude
const message = await client.messages.create({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  messages: [
    {
      role: 'user',
      content: [
        {
          type: 'image',
          source: {
            type: 'base64',
            media_type: 'image/jpeg',
            data: imageData,
          },
        },
        {
          type: 'text',
          text: 'Describe this image in detail',
        },
      ],
    },
  ],
});

console.log(message.content[0].text);
typescript
import * as fs from 'fs';

// 加载图片
const imageData = fs.readFileSync('image.jpg').toString('base64');

// 将图片发送给Claude
const message = await client.messages.create({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  messages: [
    {
      role: 'user',
      content: [
        {
          type: 'image',
          source: {
            type: 'base64',
            media_type: 'image/jpeg',
            data: imageData,
          },
        },
        {
          type: 'text',
          text: '详细描述这张图片',
        },
      ],
    },
  ],
});

console.log(message.content[0].text);

Prompt Caching (Beta)

提示词缓存（测试版）

Reduce costs by caching repetitive prompt content.
通过缓存重复的提示词内容降低成本。

Python - Prompt Caching

Python - 提示词缓存

python
undefined
python
undefined

Cache system prompt and long context

缓存系统提示词和长上下文

message = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=1024, system=[ { "type": "text", "text": "You are an expert Python developer...", "cache_control": {"type": "ephemeral"} } ], messages=[ { "role": "user", "content": "How do I use async/await?" } ] )
message = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=1024, system=[ { "type": "text", "text": "你是一位Python开发专家...", "cache_control": {"type": "ephemeral"} } ], messages=[ { "role": "user", "content": "如何使用async/await？" } ] )

Subsequent requests reuse cached system prompt

后续请求复用缓存的系统提示词

undefined
undefined

TypeScript - Prompt Caching

TypeScript - 提示词缓存

typescript
const message = await client.messages.create({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  system: [
    {
      type: 'text',
      text: 'You are an expert TypeScript developer...',
      cache_control: { type: 'ephemeral' },
    },
  ],
  messages: [
    { role: 'user', content: 'How do I use async/await?' },
  ],
});
Caching Benefits:
  • Reduces latency for repeated content
  • Lowers costs (cached tokens charged at reduced rate)
  • Useful for long system prompts, documentation, examples
typescript
const message = await client.messages.create({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  system: [
    {
      type: 'text',
      text: '你是一位TypeScript开发专家...',
      cache_control: { type: 'ephemeral' },
    },
  ],
  messages: [
    { role: 'user', content: '如何使用async/await？' },
  ],
});
缓存优势：
  • 减少重复内容的响应延迟
  • 降低成本（缓存的tokens按优惠费率收费）
  • 适用于长系统提示词、文档、示例场景

FastAPI Integration (Python)

FastAPI集成（Python）

python
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import anthropic
import os

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

class ChatRequest(BaseModel):
    message: str
    stream: bool = False

@app.post("/chat")
async def chat(request: ChatRequest):
    try:
        if request.stream:
            # Streaming response
            async def generate():
                async with client.messages.stream(
                    model="claude-3-5-sonnet-20241022",
                    max_tokens=1024,
                    messages=[{"role": "user", "content": request.message}]
                ) as stream:
                    async for text in stream.text_stream:
                        yield text

            return StreamingResponse(generate(), media_type="text/plain")
        else:
            # Non-streaming response
            message = client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1024,
                messages=[{"role": "user", "content": request.message}]
            )
            return {"response": message.content[0].text}

    except anthropic.APIError as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/chat/tools")
async def chat_with_tools(request: ChatRequest):
    tools = [
        {
            "name": "search_database",
            "description": "Search the knowledge database",
            "input_schema": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"}
                },
                "required": ["query"]
            }
        }
    ]

    message = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        tools=tools,
        messages=[{"role": "user", "content": request.message}]
    )

    return {"response": message.content, "stop_reason": message.stop_reason}
python
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import anthropic
import os

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

class ChatRequest(BaseModel):
    message: str
    stream: bool = False

@app.post("/chat")
async def chat(request: ChatRequest):
    try:
        if request.stream:
            # 流式响应
            async def generate():
                async with client.messages.stream(
                    model="claude-3-5-sonnet-20241022",
                    max_tokens=1024,
                    messages=[{"role": "user", "content": request.message}]
                ) as stream:
                    async for text in stream.text_stream:
                        yield text

            return StreamingResponse(generate(), media_type="text/plain")
        else:
            # 非流式响应
            message = client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1024,
                messages=[{"role": "user", "content": request.message}]
            )
            return {"response": message.content[0].text}

    except anthropic.APIError as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/chat/tools")
async def chat_with_tools(request: ChatRequest):
    tools = [
        {
            "name": "search_database",
            "description": "搜索知识库",
            "input_schema": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"}
                },
                "required": ["query"]
            }
        }
    ]

    message = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        tools=tools,
        messages=[{"role": "user", "content": request.message}]
    )

    return {"response": message.content, "stop_reason": message.stop_reason}

Express Integration (TypeScript)

Express集成（TypeScript）

typescript
import express from 'express';
import Anthropic from '@anthropic-ai/sdk';

const app = express();
app.use(express.json());

const client = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY,
});

interface ChatRequest {
  message: string;
  stream?: boolean;
}

app.post('/chat', async (req, res) => {
  const { message, stream }: ChatRequest = req.body;

  try {
    if (stream) {
      // Streaming response
      res.setHeader('Content-Type', 'text/plain');
      res.setHeader('Transfer-Encoding', 'chunked');

      const streamResponse = await client.messages.stream({
        model: 'claude-3-5-sonnet-20241022',
        max_tokens: 1024,
        messages: [{ role: 'user', content: message }],
      });

      for await (const chunk of streamResponse) {
        if (chunk.type === 'content_block_delta' &&
            chunk.delta.type === 'text_delta') {
          res.write(chunk.delta.text);
        }
      }
      res.end();
    } else {
      // Non-streaming response
      const response = await client.messages.create({
        model: 'claude-3-5-sonnet-20241022',
        max_tokens: 1024,
        messages: [{ role: 'user', content: message }],
      });

      res.json({ response: response.content[0].text });
    }
  } catch (error) {
    if (error instanceof Anthropic.APIError) {
      res.status(500).json({ error: error.message });
    } else {
      res.status(500).json({ error: 'Internal server error' });
    }
  }
});

app.listen(3000, () => {
  console.log('Server running on port 3000');
});
typescript
import express from 'express';
import Anthropic from '@anthropic-ai/sdk';

const app = express();
app.use(express.json());

const client = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY,
});

interface ChatRequest {
  message: string;
  stream?: boolean;
}

app.post('/chat', async (req, res) => {
  const { message, stream }: ChatRequest = req.body;

  try {
    if (stream) {
      // 流式响应
      res.setHeader('Content-Type', 'text/plain');
      res.setHeader('Transfer-Encoding', 'chunked');

      const streamResponse = await client.messages.stream({
        model: 'claude-3-5-sonnet-20241022',
        max_tokens: 1024,
        messages: [{ role: 'user', content: message }],
      });

      for await (const chunk of streamResponse) {
        if (chunk.type === 'content_block_delta' &&
            chunk.delta.type === 'text_delta') {
          res.write(chunk.delta.text);
        }
      }
      res.end();
    } else {
      // 非流式响应
      const response = await client.messages.create({
        model: 'claude-3-5-sonnet-20241022',
        max_tokens: 1024,
        messages: [{ role: 'user', content: message }],
      });

      res.json({ response: response.content[0].text });
    }
  } catch (error) {
    if (error instanceof Anthropic.APIError) {
      res.status(500).json({ error: error.message });
    } else {
      res.status(500).json({ error: '内部服务器错误' });
    }
  }
});

app.listen(3000, () => {
  console.log('服务器运行在3000端口');
});

Error Handling & Retries

错误处理与重试

Python - Error Handling

Python - 错误处理

python
from anthropic import (
    APIError,
    APIConnectionError,
    RateLimitError,
    APITimeoutError
)
import time

def chat_with_retry(message_content: str, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            message = client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1024,
                messages=[{"role": "user", "content": message_content}]
            )
            return message.content[0].text

        except RateLimitError as e:
            if attempt < max_retries - 1:
                # Exponential backoff
                wait_time = 2 ** attempt
                print(f"Rate limit hit, waiting {wait_time}s...")
                time.sleep(wait_time)
            else:
                raise

        except APIConnectionError as e:
            if attempt < max_retries - 1:
                print(f"Connection error, retrying...")
                time.sleep(1)
            else:
                raise

        except APITimeoutError as e:
            if attempt < max_retries - 1:
                print(f"Timeout, retrying...")
                time.sleep(2)
            else:
                raise

        except APIError as e:
            # Don't retry on general API errors
            print(f"API error: {e}")
            raise
python
from anthropic import (
    APIError,
    APIConnectionError,
    RateLimitError,
    APITimeoutError
)
import time

def chat_with_retry(message_content: str, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            message = client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1024,
                messages=[{"role": "user", "content": message_content}]
            )
            return message.content[0].text

        except RateLimitError as e:
            if attempt < max_retries - 1:
                # 指数退避
                wait_time = 2 ** attempt
                print(f"触发速率限制，等待{wait_time}秒...")
                time.sleep(wait_time)
            else:
                raise

        except APIConnectionError as e:
            if attempt < max_retries - 1:
                print(f"连接错误，正在重试...")
                time.sleep(1)
            else:
                raise

        except APITimeoutError as e:
            if attempt < max_retries - 1:
                print(f"请求超时，正在重试...")
                time.sleep(2)
            else:
                raise

        except APIError as e:
            # 通用API错误不重试
            print(f"API错误: {e}")
            raise

TypeScript - Error Handling

TypeScript - 错误处理

typescript
import Anthropic from '@anthropic-ai/sdk';

async function chatWithRetry(
  messageContent: string,
  maxRetries: number = 3
): Promise<string> {
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      const message = await client.messages.create({
        model: 'claude-3-5-sonnet-20241022',
        max_tokens: 1024,
        messages: [{ role: 'user', content: messageContent }],
      });

      return message.content[0].text;
    } catch (error) {
      if (error instanceof Anthropic.RateLimitError) {
        if (attempt < maxRetries - 1) {
          const waitTime = Math.pow(2, attempt) * 1000;
          console.log(`Rate limit hit, waiting ${waitTime}ms...`);
          await new Promise(resolve => setTimeout(resolve, waitTime));
        } else {
          throw error;
        }
      } else if (error instanceof Anthropic.APIConnectionError) {
        if (attempt < maxRetries - 1) {
          console.log('Connection error, retrying...');
          await new Promise(resolve => setTimeout(resolve, 1000));
        } else {
          throw error;
        }
      } else {
        // Don't retry on other errors
        throw error;
      }
    }
  }

  throw new Error('Max retries exceeded');
}
typescript
import Anthropic from '@anthropic-ai/sdk';

async function chatWithRetry(
  messageContent: string,
  maxRetries: number = 3
): Promise<string> {
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      const message = await client.messages.create({
        model: 'claude-3-5-sonnet-20241022',
        max_tokens: 1024,
        messages: [{ role: 'user', content: messageContent }],
      });

      return message.content[0].text;
    } catch (error) {
      if (error instanceof Anthropic.RateLimitError) {
        if (attempt < maxRetries - 1) {
          const waitTime = Math.pow(2, attempt) * 1000;
          console.log(`触发速率限制，等待${waitTime}毫秒...`);
          await new Promise(resolve => setTimeout(resolve, waitTime));
        } else {
          throw error;
        }
      } else if (error instanceof Anthropic.APIConnectionError) {
        if (attempt < maxRetries - 1) {
          console.log('连接错误，正在重试...');
          await new Promise(resolve => setTimeout(resolve, 1000));
        } else {
          throw error;
        }
      } else {
        // 其他错误不重试
        throw error;
      }
    }
  }

  throw new Error('已达到最大重试次数');
}

Token Counting & Cost Management

Token统计与成本管理

Python - Token Counting

Python - Token统计

python
undefined
python
undefined

Get token usage from response

从响应中获取Token使用情况

message = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=1024, messages=[{"role": "user", "content": "Hello!"}] )
print(f"Input tokens: {message.usage.input_tokens}") print(f"Output tokens: {message.usage.output_tokens}")
message = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=1024, messages=[{"role": "user", "content": "你好！"}] )
print(f"输入Token数: {message.usage.input_tokens}") print(f"输出Token数: {message.usage.output_tokens}")

Calculate cost (example rates)

计算成本（示例费率）

INPUT_COST_PER_1K = 0.003 # $3 per million tokens OUTPUT_COST_PER_1K = 0.015 # $15 per million tokens
input_cost = (message.usage.input_tokens / 1000) * INPUT_COST_PER_1K output_cost = (message.usage.output_tokens / 1000) * OUTPUT_COST_PER_1K total_cost = input_cost + output_cost
print(f"Total cost: ${total_cost:.6f}")
undefined
INPUT_COST_PER_1K = 0.003 # 每百万Token3美元 OUTPUT_COST_PER_1K = 0.015 # 每百万Token15美元
input_cost = (message.usage.input_tokens / 1000) * INPUT_COST_PER_1K output_cost = (message.usage.output_tokens / 1000) * OUTPUT_COST_PER_1K total_cost = input_cost + output_cost
print(f"总成本: ${total_cost:.6f}")
undefined

TypeScript - Token Counting

TypeScript - Token统计

typescript
const message = await client.messages.create({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  messages: [{ role: 'user', content: 'Hello!' }],
});

console.log(`Input tokens: ${message.usage.input_tokens}`);
console.log(`Output tokens: ${message.usage.output_tokens}`);

// Calculate cost
const INPUT_COST_PER_1K = 0.003;
const OUTPUT_COST_PER_1K = 0.015;

const inputCost = (message.usage.input_tokens / 1000) * INPUT_COST_PER_1K;
const outputCost = (message.usage.output_tokens / 1000) * OUTPUT_COST_PER_1K;
const totalCost = inputCost + outputCost;

console.log(`Total cost: $${totalCost.toFixed(6)}`);
typescript
const message = await client.messages.create({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  messages: [{ role: 'user', content: '你好！' }],
});

console.log(`输入Token数: ${message.usage.input_tokens}`);
console.log(`输出Token数: ${message.usage.output_tokens}`);

// 计算成本
const INPUT_COST_PER_1K = 0.003;
const OUTPUT_COST_PER_1K = 0.015;

const inputCost = (message.usage.input_tokens / 1000) * INPUT_COST_PER_1K;
const outputCost = (message.usage.output_tokens / 1000) * OUTPUT_COST_PER_1K;
const totalCost = inputCost + outputCost;

console.log(`总成本: $${totalCost.toFixed(6)}`);

Best Practices

最佳实践

Temperature & Parameters

温度参数与采样设置

python
undefined
python
undefined

Low temperature (0.0-0.3) for factual, deterministic responses

低温度（0.0-0.3）：生成事实性、确定性的响应

message = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=1024, temperature=0.1, # More focused messages=[{"role": "user", "content": "What is 2+2?"}] )
message = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=1024, temperature=0.1, # 更聚焦 messages=[{"role": "user", "content": "2+2等于多少？"}] )

Higher temperature (0.7-1.0) for creative responses

高温度（0.7-1.0）：生成创造性的响应

message = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=2048, temperature=0.9, # More creative messages=[{"role": "user", "content": "Write a creative story"}] )
message = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=2048, temperature=0.9, # 更具创造性 messages=[{"role": "user", "content": "写一个有创意的故事"}] )

Top-p (nucleus sampling)

Top-p（核采样）

message = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=1024, top_p=0.9, # Consider top 90% probability mass messages=[{"role": "user", "content": "Brainstorm ideas"}] )
undefined
message = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=1024, top_p=0.9, # 考虑概率质量前90%的词汇 messages=[{"role": "user", "content": "头脑风暴一些创意"}] )
undefined

Rate Limiting Strategies

速率限制策略

python
from datetime import datetime, timedelta
from collections import deque

class RateLimiter:
    def __init__(self, max_requests: int, time_window: int):
        self.max_requests = max_requests
        self.time_window = time_window  # seconds
        self.requests = deque()

    def can_proceed(self) -> bool:
        now = datetime.now()
        cutoff = now - timedelta(seconds=self.time_window)

        # Remove old requests
        while self.requests and self.requests[0] < cutoff:
            self.requests.popleft()

        return len(self.requests) < self.max_requests

    def add_request(self):
        self.requests.append(datetime.now())
python
from datetime import datetime, timedelta
from collections import deque

class RateLimiter:
    def __init__(self, max_requests: int, time_window: int):
        self.max_requests = max_requests
        self.time_window = time_window  # 秒
        self.requests = deque()

    def can_proceed(self) -> bool:
        now = datetime.now()
        cutoff = now - timedelta(seconds=self.time_window)

        # 移除过期的请求记录
        while self.requests and self.requests[0] < cutoff:
            self.requests.popleft()

        return len(self.requests) < self.max_requests

    def add_request(self):
        self.requests.append(datetime.now())

Usage: 50 requests per minute

使用示例：每分钟最多50次请求

limiter = RateLimiter(max_requests=50, time_window=60)
if limiter.can_proceed(): limiter.add_request() message = client.messages.create(...) else: print("Rate limit reached, waiting...")
undefined
limiter = RateLimiter(max_requests=50, time_window=60)
if limiter.can_proceed(): limiter.add_request() message = client.messages.create(...) else: print("已达到速率限制，请等待...")
undefined

Conversation Management

对话管理

python
undefined
python
undefined

Multi-turn conversation

多轮对话

conversation = []
def chat(user_message: str): # Add user message conversation.append({"role": "user", "content": user_message})
# Send to Claude
message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=conversation
)

# Add assistant response
conversation.append({
    "role": "assistant",
    "content": message.content
})

return message.content[0].text
conversation = []
def chat(user_message: str): # 添加用户消息 conversation.append({"role": "user", "content": user_message})
# 发送给Claude
message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=conversation
)

# 添加助手响应
conversation.append({
    "role": "assistant",
    "content": message.content
})

return message.content[0].text

Multi-turn usage

多轮对话使用示例

response1 = chat("What is Python?") response2 = chat("Can you show me an example?") response3 = chat("Explain the example in detail")

---
response1 = chat("什么是Python？") response2 = chat("可以给我举个例子吗？") response3 = chat("详细解释这个例子")

---

Production Patterns

生产环境模式

Connection Pooling & Timeouts

连接池与超时设置

python
undefined
python
undefined

Configure client with custom timeout

配置带自定义超时的客户端

client = anthropic.Anthropic( api_key=os.environ.get("ANTHROPIC_API_KEY"), timeout=60.0, # 60 second timeout max_retries=2, )
client = anthropic.Anthropic( api_key=os.environ.get("ANTHROPIC_API_KEY"), timeout=60.0, # 60秒超时 max_retries=2, )

For async operations

异步操作客户端

async_client = anthropic.AsyncAnthropic( api_key=os.environ.get("ANTHROPIC_API_KEY"), timeout=60.0, )
undefined
async_client = anthropic.AsyncAnthropic( api_key=os.environ.get("ANTHROPIC_API_KEY"), timeout=60.0, )
undefined

Logging & Monitoring

日志与监控

python
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def monitored_chat(user_message: str):
    start_time = time.time()

    try:
        message = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=[{"role": "user", "content": user_message}]
        )

        duration = time.time() - start_time

        logger.info(
            f"Chat completed - "
            f"Duration: {duration:.2f}s, "
            f"Input tokens: {message.usage.input_tokens}, "
            f"Output tokens: {message.usage.output_tokens}"
        )

        return message.content[0].text

    except Exception as e:
        logger.error(f"Chat failed: {e}")
        raise
python
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def monitored_chat(user_message: str):
    start_time = time.time()

    try:
        message = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=[{"role": "user", "content": user_message}]
        )

        duration = time.time() - start_time

        logger.info(
            f"对话完成 - "
            f"耗时: {duration:.2f}秒, "
            f"输入Token数: {message.usage.input_tokens}, "
            f"输出Token数: {message.usage.output_tokens}"
        )

        return message.content[0].text

    except Exception as e:
        logger.error(f"对话失败: {e}")
        raise

Environment-Based Configuration

基于环境变量的配置

python
import os
from typing import Optional

class Config:
    ANTHROPIC_API_KEY: str = os.getenv("ANTHROPIC_API_KEY", "")
    MODEL: str = os.getenv("ANTHROPIC_MODEL", "claude-3-5-sonnet-20241022")
    MAX_TOKENS: int = int(os.getenv("MAX_TOKENS", "1024"))
    TEMPERATURE: float = float(os.getenv("TEMPERATURE", "0.7"))
    TIMEOUT: float = float(os.getenv("API_TIMEOUT", "60.0"))

    @classmethod
    def validate(cls):
        if not cls.ANTHROPIC_API_KEY:
            raise ValueError("ANTHROPIC_API_KEY not set")
python
import os
from typing import Optional

class Config:
    ANTHROPIC_API_KEY: str = os.getenv("ANTHROPIC_API_KEY", "")
    MODEL: str = os.getenv("ANTHROPIC_MODEL", "claude-3-5-sonnet-20241022")
    MAX_TOKENS: int = int(os.getenv("MAX_TOKENS", "1024"))
    TEMPERATURE: float = float(os.getenv("TEMPERATURE", "0.7"))
    TIMEOUT: float = float(os.getenv("API_TIMEOUT", "60.0"))

    @classmethod
    def validate(cls):
        if not cls.ANTHROPIC_API_KEY:
            raise ValueError("未设置ANTHROPIC_API_KEY")

Initialize client with config

使用配置初始化客户端

Config.validate() client = anthropic.Anthropic( api_key=Config.ANTHROPIC_API_KEY, timeout=Config.TIMEOUT, )

---
Config.validate() client = anthropic.Anthropic( api_key=Config.ANTHROPIC_API_KEY, timeout=Config.TIMEOUT, )

---

Available Models

可用模型

ModelContext WindowBest For
claude-3-5-sonnet-20241022200K tokensGeneral purpose, reasoning, code
claude-3-5-haiku-20241022200K tokensFast responses, cost-effective
claude-3-opus-20240229200K tokensComplex tasks, highest capability
Recommended: 
claude-3-5-sonnet-20241022
for best balance of speed, cost, and capability.
模型上下文窗口最佳适用场景
claude-3-5-sonnet-20241022200K tokens通用场景、推理、代码开发
claude-3-5-haiku-20241022200K tokens快速响应、高性价比
claude-3-opus-20240229200K tokens复杂任务、最高能力要求
推荐模型： 
claude-3-5-sonnet-20241022
，在速度、成本和能力之间达到最佳平衡。

Common Pitfalls

常见陷阱

  1. Not handling tool use loops: Always check 
    stop_reason
    and handle tool use iteratively
  2. Exceeding max_tokens: Set appropriate limits based on expected response length
  3. Missing error handling: Always wrap API calls in try/catch with specific error types
  4. Ignoring rate limits: Implement exponential backoff for production systems
  5. Hardcoding API keys: Always use environment variables
  6. Not monitoring token usage: Track costs and usage in production
  7. Blocking operations: Use async clients for high-throughput applications
  1. 未处理工具调用循环：始终检查
    stop_reason
    并迭代处理工具调用
  2. 超过max_tokens限制：根据预期响应长度设置合适的限制
  3. 缺少错误处理：始终将API调用包裹在try/catch中，并处理特定错误类型
  4. 忽略速率限制：在生产系统中实现指数退避策略
  5. 硬编码API密钥：始终使用环境变量存储API密钥
  6. 未监控Token使用情况：在生产环境中跟踪成本和使用量
  7. 阻塞操作：高吞吐量应用使用异步客户端

Additional Resources

额外资源