Linux环境下Claude API代码实战：从零搭建智能对话系统

1次阅读

没有评论

共计 2586 个字符，预计需要花费 7 分钟才能阅读完成。

Claude 是 Anthropic 推出的智能对话 API（Application Programming Interface），具备以下核心能力：

多轮对话管理 ：支持上下文记忆的连贯对话
流式响应 （Streaming Response）：实现逐字返回的类真人交互体验
内容安全过滤 ：内置敏感词和违规内容检测机制

典型适用场景包括：

智能客服系统
内容创作助手
编程代码补全
知识问答引擎

优点：
开发简单，通用 HTTP 工具即可调试
对前端友好，浏览器可直接调用
缺点：
每次请求需重建连接
头信息（Header）冗余量大

优点：
二进制传输效率高
支持双向流式通信
缺点：
需要预生成桩代码（Stub）
调试工具链复杂

对于大多数应用场景，推荐优先使用 REST 方式。本文示例均基于 REST 实现。

# 确保 Python3.8+ 环境
sudo apt update && sudo apt install python3.8 python3-pip

# 安装必要工具
pip install requests python-dotenv

# 在~/.bashrc 添加
export CLAUDE_API_KEY='your_api_key_here'

# config.ini 示例
[auth]
api_key = your_api_key_here

生产环境建议结合 AWS KMS 或 HashiCorp Vault 等专业密钥管理服务。

import os
import requests
from time import sleep
from dotenv import load_dotenv

load_dotenv()  # 加载.env 文件

class ClaudeAPIClient:
    def __init__(self):
        self.base_url = "https://api.anthropic.com/v1"
        self.api_key = os.getenv("CLAUDE_API_KEY")
        self.max_retries = 3
        self.retry_delay = 2  # 秒

    def send_message(self, prompt, model="claude-2"):
        headers = {
            "Content-Type": "application/json",
            "X-API-Key": self.api_key
        }

        payload = {
            "prompt": prompt,
            "model": model,
            "max_tokens_to_sample": 1000
        }

        for attempt in range(self.max_retries):
            try:
                response = requests.post(f"{self.base_url}/complete",
                    headers=headers,
                    json=payload,
                    timeout=10
                )
                response.raise_for_status()
                return response.json()
            except requests.exceptions.RequestException as e:
                if attempt == self.max_retries - 1:
                    raise
                sleep(self.retry_delay * (attempt + 1))

# 使用示例
if __name__ == "__main__":
    client = ClaudeAPIClient()
    try:
        result = client.send_message("\n\nHuman: 你好 \n\nAssistant:")
        print(result["completion"])
    except Exception as e:
        print(f"API 调用失败: {str(e)}")

def build_context(messages):
    """ 构造多轮对话上下文

    Args:
        messages: 消息历史列表，格式为 [("user", "内容"), ("assistant", "回复")]
    """return"\n\n".join(f"{role.capitalize()}: {content}" 
        for role, content in messages
    )

import json

def handle_stream_response(response):
    """处理服务端推送的流数据"""
    for line in response.iter_lines():
        if line:
            decoded = line.decode('utf-8')
            if decoded.startswith('data:'):
                try:
                    yield json.loads(decoded[5:])
                except json.JSONDecodeError:
                    continue

首次连接超时：5-10 秒
整体请求超时：模型响应时间的 2 倍
流式响应：按需设置 60-300 秒

def sanitize_input(text):
    """基础敏感词过滤"""
    blacklist = ["信用卡", "密码", "CVV"]
    for word in blacklist:
        if word in text:
            raise ValueError("输入包含敏感词")
    return text

from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=60)
def safe_api_call():
    # 包装 API 调用逻辑
    return client.send_message(prompt)

在 4 核 8G 的 EC2 实例上测试结果：