SDK Python của Anthropic: cài đặt, thiết lập base_url, truyền dữ liệu dạng luồng và xử lý lỗi

Thư mục anthropic là gói ứng dụng khách Python chính thức của Anthropic dành cho Claude API. Gói này xử lý xác thực, thử lại, truyền luồng và các lỗi được định kiểu, đồng thời hoạt động mà không cần sửa đổi với mọi điểm cuối tương thích với Anthropic: thiết lập base_url (hoặc ANTHROPIC_BASE_URL) và phần mã còn lại của bạn giữ nguyên.

Cài đặt

pip install anthropic

Các bản phát hành 1.x chạy trên httpx2 và yêu cầu Python 3.10 trở lên. Nếu bạn truyền ứng dụng khách HTTP riêng, hãy dùng DefaultHttpxClient của SDK để giữ nguyên thời gian chờ và giới hạn kết nối mặc định.

Tạo ứng dụng khách

import anthropic

client = anthropic.Anthropic(
    api_key="your_key",   # default: ANTHROPIC_API_KEY
    base_url="https://aiprimetech.io",    # default: ANTHROPIC_BASE_URL, else https://api.anthropic.com
)

base_url là địa chỉ gốc của máy chủ. SDK tự nối thêm /v1/messages , vì vậy https://aiprimetech.io/v1 sẽ tạo ra /v1/v1/messages và một 404 page not found. Khi cả hai biến đều được thiết lập trong môi trường, anthropic.Anthropic() không cần bất kỳ đối số nào:

export ANTHROPIC_API_KEY=your_key
export ANTHROPIC_BASE_URL=https://aiprimetech.io

Gửi tin nhắn

message = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Summarise TCP slow start in two sentences."}],
)

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

print(message.usage.input_tokens, message.usage.output_tokens)
print(message._request_id)   # log this with any failure report

content là danh sách các khối (văn bản, suy nghĩ, sử dụng công cụ), vì vậy hãy kiểm tra block.type thay vì đọc content[0].text mà không kiểm tra.

Truyền luồng phản hồi

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=16000,
    messages=[{"role": "user", "content": "Write a short story about a lighthouse."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()

Hãy dùng truyền luồng cho mọi yêu cầu có đầu vào dài, đầu ra dài hoặc giá trị lớn của max_tokens: một yêu cầu không truyền luồng phải hoàn tất trong thời gian chờ, còn luồng thì không. messages.create(..., stream=True) cung cấp trình lặp sự kiện thô nếu bạn không cần tiện ích hỗ trợ.

Ứng dụng khách bất đồng bộ

import asyncio, anthropic

client = anthropic.AsyncAnthropic(base_url="https://aiprimetech.io")
limit = asyncio.Semaphore(3)   # stay within your key's concurrency

async def ask(question: str) -> str:
    async with limit:
        msg = await client.messages.create(
            model="claude-haiku-4-5",
            max_tokens=512,
            messages=[{"role": "user", "content": question}],
        )
        return next(b.text for b in msg.content if b.type == "text")

async def main():
    answers = await asyncio.gather(*(ask(q) for q in ["What is DNS?", "What is BGP?"]))
    print(answers)

asyncio.run(main())

Semaphore rất quan trọng. Gửi hàng trăm yêu cầu cùng lúc bằng asyncio.gather là nguyên nhân phổ biến nhất gây ra lỗi Concurrency limit exceeded mà chúng tôi gặp từ các ứng dụng khách Python.

Thời gian chờ và thử lại

Mặc định: thời gian chờ 10 phút cho mỗi yêu cầu và 2 lần thử lại tự động với thời gian chờ tăng theo cấp số nhân. SDK thử lại khi gặp lỗi kết nối, 408, 409, 429 và phản hồi 5xx. Vì các yêu cầu hết thời gian chờ cũng được thử lại, tổng thời gian thực tế trong trường hợp xấu nhất bằng thời gian chờ nhân với số lần thực hiện.

client = anthropic.Anthropic(base_url="https://aiprimetech.io", timeout=60.0, max_retries=4)

# override for one call only
client.with_options(timeout=300.0).messages.create(
    model="claude-opus-5-5",
    max_tokens=4000,
    messages=[{"role": "user", "content": "Review this design doc..."}],
)

Xử lý lỗi

try:
    message = client.messages.create(model="claude-sonnet-5", max_tokens=1024,
                                      messages=[{"role": "user", "content": "Hi"}])
except anthropic.NotFoundError as e:
    print("Bad model ID or URL:", e.message)
except anthropic.RateLimitError as e:
    print("Rate limited, retry after", e.response.headers.get("retry-after"))
except anthropic.APIStatusError as e:
    print(e.status_code, e.type, e.message)
except anthropic.APIConnectionError:
    print("Network problem")

Bắt các lớp ngoại lệ cụ thể trước và APIStatusError sau cùng; một khối bắt tổng quát except che khuất sự khác biệt giữa các lỗi nên thử lại và các lỗi cần sửa mã. e.type cung cấp chuỗi loại lỗi của API, chẳng hạn như rate_limit_error hoặc overloaded_error.

ID mô hình

client.models.list() trả về các mô hình mà một điểm cuối cung cấp, để bạn có thể kiểm tra ID trước khi sử dụng. Dùng ID đúng như được ghi: không có hậu tố ngày tháng.

Khắc phục sự cố trên cổng API

Tìm hiểu thêm về lỗi phía mô hình: xem hướng dẫn về lỗi Claude API.

Câu hỏi thường gặp

AI Prime Tech là một cổng API độc lập. AI Prime Tech không liên kết với, không được Anthropic chứng thực và không phải là đại lý bán lại của Anthropic. Claude và các tên mô hình liên quan là nhãn hiệu của chủ sở hữu tương ứng.