Llm integration

LLM Integration Patterns

Anthropic Claude API

import anthropic

client = anthropic.Anthropic()  # Uses ANTHROPIC_API_KEY env var

# Basic completion
message = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Explain RLHF in 3 sentences."}]
)
print(message.content[0].text)

# System prompt + streaming
with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=2048,
    system="You are a senior data scientist. Be concise and precise.",
    messages=[{"role": "user", "content": "Review this model architecture: ..."}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

Tool Use (Function Calling)

tools = [
    {
        "name": "get_stock_price",
        "description": "Get current stock price for a ticker symbol",
        "input_schema": {
            "type": "object",
            "properties": {
                "ticker": {"type": "string", "description": "Stock ticker e.g. AAPL"},
            },
            "required": ["ticker"]
        }
    }
]

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "What's Apple's current stock price?"}]
)

if response.stop_reason == "tool_use":
    tool_use = next(b for b in response.content if b.type == "tool_use")
    result = get_stock_price(tool_use.input["ticker"])

    # Continue conversation with tool result
    final = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=512,
        tools=tools,
        messages=[
            {"role": "user", "content": "What's Apple's stock price?"},
            {"role": "assistant", "content": response.content},
            {"role": "user", "content": [{"type": "tool_result", "tool_use_id": tool_use.id, "content": str(result)}]},
        ]
    )

LangChain Chains

from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatAnthropic(model="claude-sonnet-4-6")

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a code reviewer. Identify bugs and suggest fixes."),
    ("human", "Review this code:\n\n{code}"),
])

chain = prompt | llm | StrOutputParser()

result = chain.invoke({"code": "def divide(a, b): return a / b"})

# Async streaming
async for chunk in chain.astream({"code": "..."}):
    print(chunk, end="", flush=True)

Prompt Templates

from langchain_core.prompts import PromptTemplate

# Few-shot template
few_shot_template = PromptTemplate.from_template("""
Classify the sentiment of customer reviews.

Examples:
Review: "Amazing product, works perfectly!" → Positive
Review: "Broken on arrival, terrible quality" → Negative
Review: "It's okay, nothing special" → Neutral

Review: "{review}"
Sentiment:""")

# Dynamic few-shot with example selector
from langchain_core.example_selectors import SemanticSimilarityExampleSelector
from langchain_community.vectorstores import FAISS

selector = SemanticSimilarityExampleSelector.from_examples(
    examples=training_examples,
    embeddings=embeddings,
    vectorstore_cls=FAISS,
    k=3,  # Select 3 most relevant examples
)

Caching & Cost Management

from langchain_community.cache import SQLiteCache
from langchain_core.globals import set_llm_cache

# Cache responses to avoid repeat API calls
set_llm_cache(SQLiteCache(database_path=".langchain.db"))

# Token counting before call
import anthropic
client = anthropic.Anthropic()

# Count tokens
token_count = client.messages.count_tokens(
    model="claude-sonnet-4-6",
    messages=[{"role": "user", "content": long_document}]
)
print(f"Input tokens: {token_count.input_tokens}")

# Prompt caching for repeated context (saves cost)
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    system=[{
        "type": "text",
        "text": large_system_prompt,
        "cache_control": {"type": "ephemeral"}  # Cache this prefix
    }],
    messages=[{"role": "user", "content": "Summarize section 3"}]
)

Structured Output

from pydantic import BaseModel
from langchain_anthropic import ChatAnthropic

class CodeReview(BaseModel):
    bugs: list[str]
    suggestions: list[str]
    severity: str  # "low", "medium", "high"
    score: int     # 1-10

llm = ChatAnthropic(model="claude-sonnet-4-6")
structured_llm = llm.with_structured_output(CodeReview)

review = structured_llm.invoke("Review this Python function: def foo(x): return x*x")
print(review.bugs, review.score)