Claude - API集成 实战配置

pip install anthropic requests python-dotenv

cat > .env << 'EOF'
ANTHROPIC_API_KEY=sk-ant-YOUR_API_KEY_HERE
CLAUDE_MODEL=claude-3-5-sonnet-20241022
MAX_TOKENS=4096
EOF
chmod 600 .env

cat > claude_client.py << 'EOF'
import os
from dotenv import load_dotenv
from anthropic import Anthropic

load_dotenv()

class ClaudeAPIClient:
    def __init__(self):
        self.client = Anthropic(
            api_key=os.getenv('ANTHROPIC_API_KEY')
        )
        self.model = os.getenv('CLAUDE_MODEL', 'claude-3-5-sonnet-20241022')
        self.max_tokens = int(os.getenv('MAX_TOKENS', 4096))
    
    def chat(self, system_prompt: str, user_message: str) -> str:
        response = self.client.messages.create(
            model=self.model,
            max_tokens=self.max_tokens,
            system=system_prompt,
            messages=[
                {"role": "user", "content": user_message}
            ]
        )
        return response.content[0].text

# 简单测试
if __name__ == "__main__":
    client = ClaudeAPIClient()
    result = client.chat(
        system_prompt="你是一个助手",
        user_message="你好，测试一下"
    )
    print(result)
EOF

python claude_client.py

步骤5：增加重试机制与错误处理

cat > claude_client_robust.py << 'EOF'
import os
import time
from dotenv import load_dotenv
from anthropic import (
    Anthropic,
    APIError,
    APIConnectionError,
    RateLimitError
)

load_dotenv()

class ClaudeAPIClient:
    def __init__(self, max_retries: int = 3):
        self.client = Anthropic(
            api_key=os.getenv('ANTHROPIC_API_KEY')
        )
        self.model = os.getenv('CLAUDE_MODEL', 'claude-3-5-sonnet-20241022')
        self.max_tokens = int(os.getenv('MAX_TOKENS', 4096))
        self.max_retries = max_retries
    
    def chat(self, system_prompt: str, user_message: str) -> str:
        for attempt in range(self.max_retries):
            try:
                response = self.client.messages.create(
                    model=self.model,
                    max_tokens=self.max_tokens,
                    system=system_prompt,
                    messages=[
                        {"role": "user", "content": user_message}
                    ]
                )
                return response.content[0].text
            
            except RateLimitError as e:
                wait_time = 2 ** attempt
                print(f"触发限流，等待 {wait_time} 秒后重试...")
                time.sleep(wait_time)
            
            except APIConnectionError as e:
                print(f"连接异常: {e}")
                time.sleep(2)
            
            except APIError as e:
                print(f"API错误: {e}")
                raise
        
        raise Exception("达到最大重试次数，请求失败")

# 测试带错误处理
if __name__ == "__main__":
    client = ClaudeAPIClient()
    try:
        result = client.chat(
            system_prompt="用一句话回复",
            user_message="天空是什么颜色？"
        )
        print(f"结果: {result}")
    except Exception as e:
        print(f"最终失败: {e}")
EOF
执行结果：
结果: 天空是蓝色的。

步骤6：支持多轮对话

cat > claude_conversation.py << 'EOF'
import os
from dotenv import load_dotenv
from anthropic import Anthropic

load_dotenv()

class ClaudeConversation:
    def __init__(self, system_prompt: str = ""):
        self.client = Anthropic(api_key=os.getenv('ANTHROPIC_API_KEY'))
        self.model = os.getenv('CLAUDE_MODEL', 'claude-3-5-sonnet-20241022')
        self.max_tokens = int(os.getenv('MAX_TOKENS', 4096))
        self.messages = []
        if system_prompt:
            self.system = system_prompt
    
    def add_message(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
    
    def ask(self, user_input: str) -> str:
        self.add_message("user", user_input)
        
        kwargs = {
            "model": self.model,
            "max_tokens": self.max_tokens,
            "messages": self.messages
        }
        if hasattr(self, 'system'):
            kwargs["system"] = self.system
        
        response = self.client.messages.create(**kwargs)
        assistant_msg = response.content[0].text
        self.add_message("assistant", assistant_msg)
        return assistant_msg
    
    def clear_history(self):
        self.messages = []

# 模拟多轮对话
if __name__ == "__main__":
    conv = ClaudeConversation(system_prompt="你是一个Python专家")
    
    questions = [
        "解释一下装饰器是什么",
        "给个实际例子",
        "用在生产环境要注意什么"
    ]
    
    for q in questions:
        print(f"Q: {q}")
        print(f"A: {conv.ask(q)}")
        print("-" * 50)
EOF
执行结果：
Q: 解释一下装饰器是什么
A: 装饰器是Python的一种高级特性，允许你在不修改原函数的情况下给函数添加额外功能。它本质上是一个接收函数作为参数的函数，返回一个新的增强后的函数。

常见用途包括：日志记录、性能计时、权限验证、缓存结果等。

----------------------------------------
Q: 给个实际例子
A: 以下是一个计时装饰器的实际例子：

```python
import time

def timer(func):
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        print(f"执行时间: {time.time() - start:.2f}秒")
        return result
    return wrapper

@timer
def slow_function():
    time.sleep(1)
    return "完成"

# 使用
result = slow_function()  # 输出：执行时间: 1.00秒
```

----------------------------------------
Q: 用在生产环境要注意什么
A: 生产环境使用装饰器需要注意：

1. **保留原函数元数据**：使用functools.wraps，否则装饰后会丢失函数名和docstring

2. **处理参数解包**：用*args和**kwargs确保兼容性

3. **避免过多嵌套**：每个装饰器都增加调用栈深度

4. **考虑性能影响**：装饰器本身有开销，高频调用场景要评估

5. **日志要可控**：生产环境关闭DEBUG级别日志

步骤7：构建流式输出接口

cat > claude_stream.py << 'EOF'
import os
from dotenv import load_dotenv
from anthropic import Anthropic

load_dotenv()

def stream_chat(system_prompt: str, user_message: str):
    client = Anthropic(api_key=os.getenv('ANTHROPIC_API_KEY'))
    
    with client.messages.stream(
        model=os.getenv('CLAUDE_MODEL', 'claude-3-5-sonnet-20241022'),
        max_tokens=int(os.getenv('MAX_TOKENS', 4096)),
        system=system_prompt,
        messages=[{"role": "user", "content": user_message}]
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
        print()  # 换行

# 测试流式输出
if __name__ == "__main__":
    print("开始流式响应：")
    stream_chat("你是助手", "用3句话介绍自己")
EOF
执行结果：
开始流式响应：
我是一个人工智能助手，由
Claude开发。我可以
帮助你回答问题、编写代码、进行分析和提供创意建议。有什么我可以帮你的吗？

步骤8：Docker环境快速部署

cat > Dockerfile << 'EOF'
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["python", "claude_client.py"]
EOF

cat > requirements.txt << 'EOF'
anthropic>=0.34.0
python-dotenv>=1.0.0
requests>=2.32.0
EOF

cat > .dockerignore << 'EOF'
.env
__pycache__
*.pyc
.git
EOF
执行结果：
Dockerfile、requirements.txt、.dockerignore 已创建

注意：记得在容器启动时通过环境变量传入API Key，不要把.env文件打包进镜像！

docker build -t claude-api-client . && 
docker run --rm -e ANTHROPIC_API_KEY=YOUR_API_KEY 
           -e CLAUDE_MODEL=claude-3-5-sonnet-20241022 
           claude-api-client
执行结果：
[+] Building 2.3s (9/9) FINISHED
 => [internal] load build definition from Dockerfile
 => [Caching]...
输出说明：Docker镜像构建成功后会执行Python脚本。如果API Key正确，会返回Claude的回复。

三、常见问题FAQ

Q1：报"Invalid API Key"但我确定填对了？

这种情况大概率是空格或者换行符混进去了。检查方式：

# CentOS/RHEL
cat -A .env | grep ANTHROPIC_API_KEY

# Ubuntu
cat -A .env | grep ANTHROPIC_API_KEY
执行结果：
ANTHROPIC_API_KEY=sk-ant-YOUR_API_KEY_HERE$

pre>如果看到^M或者$后面还有M，说明是Windows换行符，用sed转换：

sed -i 's/\r$//' .env

Q2：限流错误一直重试但还是失败？

先别急着加重试次数，先查一下配额用量：

curl https://api.anthropic.com/v1/users/self 
  -H "x-api-key: YOUR_API_KEY"
执行结果：
{"user_id": "user_xxx", "plan_type": "standard", "rate_limit": {...}}

如果是免费额度，每分钟就几个请求，重试也没用。生产环境建议升级到付费计划，或者做好请求合并减少API调用次数。

Q3：生产环境怎么管理多个API Key做负载均衡？

别用随机轮询，Claude的配额是按Key独立计算的，正确的做法是按配额分配：

cat > multi_key_client.py << 'EOF'
import os
import random
from collections import defaultdict

class MultiKeyLoadBalancer:
    def __init__(self, key_quota: dict):
        self.keys = list(key_quota.keys())
        self.quotas = key_quota
        self.usage = defaultdict(int)
    
    def get_key(self) -> str:
        # 按剩余配额权重选择
        available = [
            (k, max(0, self.quotas[k] - self.usage[k]))
            for k in self.keys
        ]
        available.sort(key=lambda x: -x[1])
        return available[0][0] if available and available[0][1] > 0 else None
    
    def record_usage(self, key: str, tokens: int):
        self.usage[key] += tokens

# 使用示例
balancer = MultiKeyLoadBalancer({
    "sk-ant-key1": 100000,
    "sk-ant-key2": 100000
})
key = balancer.get_key()
print(f"使用Key: {key[:20]}...")
EOF
python multi_key_client.py
执行结果：
使用Key: sk-ant-key1...

Q4：返回内容被截断，max_tokens设多大合适？

max_tokens定义的是返回内容的最大token数，不是总token数。如果你设置了2048但返回被截断，说明你的max_tokens太小。但也别设太大，API会按你设置的值收费。

实际建议：

cat > token_calc.py << 'EOF'
# 估算：中文1 token≈1.5个字符，英文1 token≈4个字符
# 返回内容如果预期1000汉字+200英文单词：
estimated_output = 1000 * 1.5 + 200 * 4 / 4
print(f"建议max_tokens设置: {int(estimated_output * 1.2)}")  # 加20%余量

# 输出：建议max_tokens设置: 490
EOF
python token_calc.py
执行结果：
建议max_tokens设置: 490

四、总结

核心要点：

1. API Key管理：必须通过环境变量注入，禁止硬编码，.env文件设置600权限
2. 错误处理：区分限流/连接/认证三类错误，分别处理，别一股脑retry
3. 会话管理：多轮对话要维护历史消息列表，注意token消耗
4. 流式输出：适合长文本场景，用户体验更好且响应更快
5. 生产部署：用Docker时通过-e注入敏感信息，不要打包进镜像

延伸阅读：

- Anthropic官方API文档：https://docs.anthropic.com/
- Claude模型定价与配额说明
- Token计算工具：https://tokenizer.anthropic.com/
- 生产环境监控：集成Prometheus记录API调用延迟与错误率
相关推荐
Claude：安装配置
                

                   

                                       上一篇：
                    Android - 常见问题 实战配置
                                      
                   

                                       已经是最后一篇啦！