Spaces:
Build error
Build error
File size: 4,871 Bytes
9c7d451 e4a41fa 9c7d451 e4a41fa 9c7d451 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 | #!/usr/bin/env python3
"""
Anthropic example for Headroom SDK.
This example shows how to use Headroom with Anthropic Claude models.
"""
import os
import tempfile
from anthropic import Anthropic
from dotenv import load_dotenv
from headroom import AnthropicProvider, HeadroomClient
# Load API key from .env.local
load_dotenv(".env.local")
# Create base Anthropic client
base_client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
# Create provider for Anthropic models
provider = AnthropicProvider()
# Use temp directory for database
db_path = os.path.join(tempfile.gettempdir(), "headroom_anthropic.db")
# Wrap with Headroom
client = HeadroomClient(
original_client=base_client,
provider=provider,
store_url=f"sqlite:///{db_path}",
default_mode="audit",
)
def example_audit_mode():
"""Example using audit mode (observe only)."""
print("=" * 50)
print("ANTHROPIC AUDIT MODE EXAMPLE")
print("=" * 50)
messages = [
{"role": "user", "content": "What's 2 + 2? Reply in one word."},
]
# In audit mode, request passes through unchanged but metrics are logged
response = client.messages.create(
model="claude-3-5-haiku-latest",
messages=messages,
max_tokens=100,
)
print(f"Response: {response.content[0].text}")
print()
def example_optimize_mode():
"""Example using optimize mode (apply transforms)."""
print("=" * 50)
print("ANTHROPIC OPTIMIZE MODE EXAMPLE")
print("=" * 50)
messages = [
{"role": "user", "content": "Search for information."},
{
"role": "assistant",
"content": [
{
"type": "tool_use",
"id": "call_1",
"name": "search",
"input": {"query": "test"},
}
],
},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "call_1",
"content": '{"results": ['
+ ",".join([f'{{"id": {i}}}' for i in range(50)])
+ "]}",
}
],
},
{"role": "assistant", "content": "I found 50 results."},
{"role": "user", "content": "Summarize them briefly."},
]
# In optimize mode, transforms are applied
response = client.messages.create(
model="claude-3-5-haiku-latest",
messages=messages,
headroom_mode="optimize",
max_tokens=100,
)
print(f"Response: {response.content[0].text}")
print()
def example_simulate_mode():
"""Example using simulate mode (preview without API call)."""
print("=" * 50)
print("ANTHROPIC SIMULATE MODE EXAMPLE")
print("=" * 50)
messages = [
{"role": "user", "content": "Search for information."},
{
"role": "assistant",
"content": [
{
"type": "tool_use",
"id": "call_1",
"name": "search",
"input": {"query": "test"},
}
],
},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "call_1",
"content": '{"results": ['
+ ",".join([f'{{"id": {i}}}' for i in range(100)])
+ "]}",
}
],
},
{"role": "assistant", "content": "I found 100 results."},
{"role": "user", "content": "Summarize them."},
]
# Simulate without calling API
plan = client.messages.simulate(
model="claude-3-5-sonnet-latest",
messages=messages,
)
print(f"Tokens before: {plan.tokens_before}")
print(f"Tokens after: {plan.tokens_after}")
print(f"Tokens saved: {plan.tokens_saved}")
print(f"Transforms applied: {plan.transforms}")
print(f"Estimated savings: {plan.estimated_savings}")
print()
def example_streaming():
"""Example of streaming with Anthropic."""
print("=" * 50)
print("ANTHROPIC STREAMING EXAMPLE")
print("=" * 50)
messages = [
{"role": "user", "content": "Count from 1 to 5. Just the numbers."},
]
# Stream with optimization
with client.messages.stream(
model="claude-3-5-haiku-latest",
messages=messages,
headroom_mode="optimize",
max_tokens=100,
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
print()
print()
if __name__ == "__main__":
# Run examples
example_audit_mode()
example_optimize_mode()
example_simulate_mode()
example_streaming()
# Clean up
client.close()
|