Anthropic Client

pip install graphiti-core[anthropic]

from graphiti_core.llm_client import AnthropicClient
from graphiti_core.llm_client.config import LLMConfig
from graphiti_core.prompts.models import Message
from pydantic import BaseModel

# Initialize client
client = AnthropicClient(
    config=LLMConfig(
        api_key="sk-ant-...",  # Or set ANTHROPIC_API_KEY env var
        model="claude-haiku-4-5-latest",
        temperature=1.0,
        max_tokens=16384
    )
)

# Define response structure
class Analysis(BaseModel):
    sentiment: str
    confidence: float
    key_themes: list[str]

# Generate structured response
messages = [
    Message(role="system", content="Analyze the sentiment of the text."),
    Message(role="user", content="I absolutely love this product!")
]

response = await client.generate_response(
    messages=messages,
    response_model=Analysis
)

# Example: Using different max_tokens strategies

# 1. Model-specific (automatic): 65536 for claude-4-5
client = AnthropicClient(
    config=LLMConfig(model="claude-sonnet-4-5-latest")
)

# 2. Instance-level: 32K for all requests
client = AnthropicClient(
    config=LLMConfig(model="claude-haiku-4-5-latest"),
    max_tokens=32000
)

# 3. Per-request: 8K for this specific request
response = await client.generate_response(
    messages=messages,
    max_tokens=8192
)

# With response_model
class PersonInfo(BaseModel):
    """Extract person information"""
    name: str
    age: int
    location: str

response = await client.generate_response(
    messages=messages,
    response_model=PersonInfo
)
# Returns: {'name': '...', 'age': 30, 'location': '...'}

# Without response_model (generic JSON)
response = await client.generate_response(
    messages=messages
)
# Returns: any JSON object

# For PersonInfo model:
tool = {
    'name': 'PersonInfo',
    'description': 'Extract person information',
    'input_schema': {
        'type': 'object',
        'properties': {
            'name': {'type': 'string'},
            'age': {'type': 'integer'},
            'location': {'type': 'string'}
        },
        'required': ['name', 'age', 'location']
    }
}

from graphiti_core.llm_client.errors import RateLimitError

try:
    response = await client.generate_response(messages=messages)
except RateLimitError as e:
    print(f"Rate limited: {e}")
    # No automatic retry - implement backoff in your code

from graphiti_core.llm_client.errors import RefusalError

try:
    response = await client.generate_response(messages=messages)
except RefusalError as e:
    print(f"Content policy violation: {e}")
    # No retry - content was rejected

# Retry flow for validation errors:
# 1. First attempt fails validation
# 2. Error context appended to messages
# 3. Retry with guidance
# 4. Up to 2 total retries before raising exception

client = AnthropicClient()

response = await client.generate_response(
    messages=messages,
    prompt_name="entity_extraction"
)

# Check usage
usage = client.token_tracker.get_usage()
print(f"Input tokens: {usage['input_tokens']}")
print(f"Output tokens: {usage['output_tokens']}")
print(f"Total tokens: {usage['total_tokens']}")

# Usage by prompt name
prompt_usage = client.token_tracker.get_usage_by_prompt("entity_extraction")

# Extracts JSON from responses like:
# "Here's the information: {\"name\": \"John\", \"age\": 30}"

# Looks for first '{' to last '}'
json_start = text.find('{')
json_end = text.rfind('}') + 1
json_str = text[json_start:json_end]
return json.loads(json_str)

from graphiti_core.llm_client import AnthropicClient
from graphiti_core.prompts.models import Message

client = AnthropicClient(
    config=LLMConfig(model="claude-sonnet-4-5-latest")
)

messages = [
    Message(role="system", content="You are a helpful assistant."),
    Message(role="user", content="What's the capital of France?"),
]

# First response
response1 = await client.generate_response(messages=messages)
print(response1)  # {'content': 'The capital of France is Paris.'}

# Continue conversation
messages.append(Message(role="user", content="What's its population?"))
response2 = await client.generate_response(messages=messages)

# Automatic model selection
response = await client.generate_response(
    messages=messages,
    model_size=ModelSize.small  # Uses small_model from config
)