跳到主要内容

流式输出

import anthropic

client = anthropic.Anthropic(api_key="your-api-key", base_url="https://www.cheapertoken.work")

with client.messages.stream(
model="claude-opus-4-7",
max_tokens=1024,
messages=[{"role": "user", "content": "写一首关于冬天的诗"}],
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)

获取最终结果

import anthropic

client = anthropic.Anthropic(api_key="your-api-key", base_url="https://www.cheapertoken.work")

with client.messages.stream(
model="claude-opus-4-7",
max_tokens=1024,
messages=[{"role": "user", "content": "你好"}],
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)

# 流结束后获取完整消息对象
final_message = stream.get_final_message()
print(f"\n总 Token 数: {final_message.usage.input_tokens + final_message.usage.output_tokens}")

异步流式输出

import asyncio
import anthropic

client = anthropic.AsyncAnthropic(api_key="your-api-key", base_url="https://www.cheapertoken.work")

async def main():
async with client.messages.stream(
model="claude-opus-4-7",
max_tokens=1024,
messages=[{"role": "user", "content": "写一首诗"}],
) as stream:
async for text in stream.text_stream:
print(text, end="", flush=True)

asyncio.run(main())