Phase 6: AI Integration

pip install anthropic

ANTHROPIC_API_KEY=sk-ant-api03-...

import anthropic
from dotenv import load_dotenv
load_dotenv()

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    system="You are a concise Python tutor. Answer in 2-3 sentences max.",
    messages=[
        {"role": "user", "content": "What is a generator in Python?"}
    ]
)

# The response object — know its structure
print(response.content[0].text)    # the actual text response
print(response.model)              # which model responded
print(response.stop_reason)        # "end_turn" or "max_tokens"
print(response.usage.input_tokens)  # tokens you sent
print(response.usage.output_tokens) # tokens in the response

import anthropic
from dataclasses import dataclass, field

@dataclass
class Conversation:
    """Manages a stateful conversation with Claude."""
    system:   str              = "You are a helpful assistant."
    model:    str              = "claude-sonnet-4-5"
    history:  list             = field(default_factory=list)
    _client:  anthropic.Anthropic = field(
        default_factory=anthropic.Anthropic, init=False, repr=False
    )

    def chat(self, message: str) -> str:
        # Add the user message to history
        self.history.append({"role": "user", "content": message})

        response = self._client.messages.create(
            model=self.model,
            max_tokens=2048,
            system=self.system,
            messages=self.history,  # send FULL history every time
        )

        reply = response.content[0].text
        # Add Claude's reply to history for the next turn
        self.history.append({"role": "assistant", "content": reply})
        return reply

    def reset(self) -> None:
        self.history.clear()

    def token_count(self) -> int:
        # Rough estimate: 4 chars per token
        total_chars = sum(len(m["content"]) for m in self.history)
        return total_chars // 4


# Usage
conv = Conversation(system="You are a Python expert. Be concise.")
print(conv.chat("What's the difference between a list and a tuple?"))
print(conv.chat("When would I use one over the other?"))
print(conv.chat("Give me a real example from a web API."))
# All three questions share context — Claude remembers the conversation

pip install openai

from openai  import OpenAI
from dotenv  import load_dotenv
load_dotenv()
# Reads OPENAI_API_KEY from environment automatically

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o",
    max_tokens=1024,
    messages=[
        {"role": "system",  "content": "You are a concise Python tutor."},
        {"role": "user",    "content": "What is a generator?"}
    ]
)

# Different attribute path than Anthropic
text   = response.choices[0].message.content
reason = response.choices[0].finish_reason   # "stop" or "length"
tokens = response.usage.total_tokens

from dataclasses import dataclass
from enum        import Enum
from typing      import Literal

class Provider(Enum):
    ANTHROPIC = "anthropic"
    OPENAI    = "openai"

def chat(
    prompt:   str,
    system:   str  = "You are a helpful assistant.",
    provider: Provider = Provider.ANTHROPIC,
    model:    str  = None,
    max_tokens: int = 1024,
    temperature: float = 0.7,
) -> str:
    """Single function to call any LLM provider."""

    if provider == Provider.ANTHROPIC:
        import anthropic
        model = model or "claude-sonnet-4-5"
        client = anthropic.Anthropic()
        response = client.messages.create(
            model=model, max_tokens=max_tokens,
            system=system,
            messages=[{"role": "user", "content": prompt}]
        )
        return response.content[0].text

    elif provider == Provider.OPENAI:
        from openai import OpenAI
        model = model or "gpt-4o"
        client = OpenAI()
        response = client.chat.completions.create(
            model=model, max_tokens=max_tokens, temperature=temperature,
            messages=[
                {"role": "system", "content": system},
                {"role": "user",   "content": prompt}
            ]
        )
        return response.choices[0].message.content

# Use the same function regardless of provider
result = chat("Explain list comprehensions", provider=Provider.ANTHROPIC)
result = chat("Explain list comprehensions", provider=Provider.OPENAI)
# Swap the provider line to compare quality and cost

import anthropic

client = anthropic.Anthropic()

# ── Method 1: stream context manager (recommended) ──
with client.messages.stream(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Write a haiku about Python."}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)  # print each token as it arrives

    # After the loop, get final usage stats
    message = stream.get_final_message()
    print(f"\n\nTokens: {message.usage.input_tokens} in, {message.usage.output_tokens} out")


# ── Method 2: collect the full streamed text ──
with client.messages.stream(
    model="claude-sonnet-4-5",
    max_tokens=2048,
    system="You are a code reviewer.",
    messages=[{"role": "user", "content": "Review this function: def add(a,b): return a+b"}]
) as stream:
    full_text = stream.get_final_text()  # waits for completion, returns string

print(full_text)

from fastapi           import FastAPI
from fastapi.responses import StreamingResponse
from pydantic          import BaseModel
import anthropic

app    = FastAPI()
client = anthropic.Anthropic()

class ChatRequest(BaseModel):
    message: str
    system:  str = "You are a helpful assistant."

def token_generator(message: str, system: str):
    """Generator that yields tokens from Claude as they arrive."""
    with client.messages.stream(
        model="claude-sonnet-4-5",
        max_tokens=2048,
        system=system,
        messages=[{"role": "user", "content": message}]
    ) as stream:
        for text in stream.text_stream:
            yield text   # yield each token to the HTTP response

@app.post("/chat/stream")
def stream_chat(req: ChatRequest):
    return StreamingResponse(
        token_generator(req.message, req.system),
        media_type="text/plain"
    )

# The client receives a stream of text chunks:
# "A" → " list" → " comprehension" → " is" → " a" → " concise" → ...
# In a React frontend: fetch with ReadableStream to display token-by-token

from string  import Template
from pathlib import Path

# ── Simple f-string templates ──
def summarize_prompt(text: str, max_words: int = 100) -> str:
    return f"""Summarize the following text in {max_words} words or fewer.
Focus on the key points. Use plain language.

TEXT:
{text}

SUMMARY:"""

def classify_prompt(text: str, categories: list[str]) -> str:
    cats = ", ".join(categories)
    return f"""Classify the following text into exactly one of these categories: {cats}

Respond with ONLY the category name. No explanation.

TEXT: {text}

CATEGORY:"""

# ── Few-shot template — examples teach the format ──
def extract_tasks_prompt(text: str) -> str:
    return f"""Extract action items from the text as a JSON array.
Each item has: "task" (string), "owner" (string or null), "due" (date string or null).

EXAMPLE INPUT:
"John needs to finish the report by Friday. Sarah will review it next week."

EXAMPLE OUTPUT:
[
  {{"task": "Finish the report", "owner": "John", "due": "Friday"}},
  {{"task": "Review the report", "owner": "Sarah", "due": "next week"}}
]

INPUT:
{text}

OUTPUT:"""

# Usage
import anthropic, json

client = anthropic.Anthropic()

def run_prompt(prompt: str, system: str = "You are a helpful assistant.",
               temperature: float = 0) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        temperature=temperature,
        system=system,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text

# Extract tasks from a meeting transcript
transcript = "Jude needs to push the new feature by Thursday. The team will demo it on Friday."
result = run_prompt(extract_tasks_prompt(transcript))
tasks  = json.loads(result)
print(tasks)
# [{"task": "Push the new feature", "owner": "Jude", "due": "Thursday"}, ...]

def analyze_code_prompt(code: str) -> str:
    return f"""Analyze the following Python code for bugs and improvements.

Think through this step by step:
1. First, understand what the code is trying to do
2. Identify any bugs or errors
3. Note any performance issues
4. Suggest improvements to readability or Pythonic style
5. Give a final summary with a priority-ordered list of changes

CODE:
```python
{code}
```

Begin your analysis:"""

# Chain-of-thought works because the model "reasons aloud" before answering
# This forces it to consider each aspect rather than jumping to a conclusion
# Use it for: code review, debugging, complex decisions, data analysis


# ── Structured output — ask for JSON explicitly ──
def extract_structured(text: str) -> dict:
    prompt = f"""Extract information from the text below.
Respond with ONLY valid JSON. No markdown, no explanation, no backticks.

Required fields:
- "name": string
- "email": string or null
- "phone": string or null
- "company": string or null

TEXT: {text}"""

    response = run_prompt(prompt, temperature=0)

    # Strip any accidental markdown code fences
    clean = response.strip().removeprefix("```json").removeprefix("```").removesuffix("```")
    return json.loads(clean)

# Step 1: You define available tools as JSON schemas
tools = [{
    "name":        "get_weather",
    "description": "Get the current weather for a city.",
    "input_schema": {
        "type": "object",
        "properties": {
            "city": {"type": "string", "description": "City name"},
        },
        "required": ["city"]
    }
}]

# Step 2: The model returns a tool_use block instead of text
# response.content[0].type == "tool_use"
# response.content[0].name == "get_weather"
# response.content[0].input == {"city": "Bossier City"}

# Step 3: You run the function and send the result back
# The model uses the result to form its final response

import anthropic, json, requests

client = anthropic.Anthropic()

# ── Define your actual Python functions ──
def get_weather(city: str) -> dict:
    # Open-Meteo: free, no API key needed
    geo = requests.get(
        "https://geocoding-api.open-meteo.com/v1/search",
        params={"name": city, "count": 1}, timeout=5
    ).json()
    lat = geo["results"][0]["latitude"]
    lon = geo["results"][0]["longitude"]
    weather = requests.get(
        "https://api.open-meteo.com/v1/forecast",
        params={"latitude": lat, "longitude": lon,
                "current": "temperature_2m,wind_speed_10m"}, timeout=5
    ).json()
    current = weather["current"]
    return {"city": city, "temp_c": current["temperature_2m"],
            "wind_kmh": current["wind_speed_10m"]}

def calculate(expression: str) -> dict:
    try:
        result = eval(expression, {"__builtins__": {}})
        return {"expression": expression, "result": result}
    except Exception as e:
        return {"error": str(e)}

# ── Tool registry — maps tool names to functions ──
TOOLS_REGISTRY = {
    "get_weather": get_weather,
    "calculate":   calculate,
}

# ── Tool schemas — what the model needs to know ──
TOOLS = [
    {
        "name": "get_weather",
        "description": "Get current temperature and wind speed for a city.",
        "input_schema": {
            "type": "object",
            "properties": {"city": {"type": "string"}},
            "required": ["city"]
        }
    },
    {
        "name": "calculate",
        "description": "Evaluate a mathematical expression. Returns the numeric result.",
        "input_schema": {
            "type": "object",
            "properties": {"expression": {"type": "string"}},
            "required": ["expression"]
        }
    }
]


def run_agent(user_message: str) -> str:
    """Run the full tool use loop until the model gives a final response."""
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=1024,
            tools=TOOLS,
            messages=messages
        )

        # If the model is done, return the text response
        if response.stop_reason == "end_turn":
            return response.content[0].text

        # The model wants to use a tool
        if response.stop_reason == "tool_use":
            # Add model's response (including tool use request) to history
            messages.append({"role": "assistant", "content": response.content})

            # Process each tool call
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    func    = TOOLS_REGISTRY[block.name]
                    result  = func(**block.input)
                    print(f"🔧 Called {block.name}({block.input}) → {result}")
                    tool_results.append({
                        "type":        "tool_result",
                        "tool_use_id": block.id,
                        "content":     json.dumps(result)
                    })

            # Send the tool results back to the model
            messages.append({"role": "user", "content": tool_results})


# Try it
answer = run_agent("What's the weather in Bossier City, and what's 15% of 87?")
print(answer)
# 🔧 Called get_weather({'city': 'Bossier City'}) → {'city': 'Bossier City', 'temp_c': 32.1, ...}
# 🔧 Called calculate({'expression': '0.15 * 87'}) → {'expression': '0.15 * 87', 'result': 13.05}
# "In Bossier City it's currently 32.1°C with wind at 14.2 km/h. 15% of 87 is 13.05."

pip install numpy  # for cosine similarity

import anthropic, json, numpy as np
from pathlib import Path

client = anthropic.Anthropic()


# ── Step 1: Chunk your documents ──
def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
    """Split text into overlapping chunks."""
    words  = text.split()
    chunks = []
    for i in range(0, len(words), chunk_size - overlap):
        chunk = " ".join(words[i : i + chunk_size])
        if chunk: chunks.append(chunk)
    return chunks


# ── Step 2: Generate embeddings ──
def get_embedding(text: str) -> list[float]:
    """Get embedding vector for a text string using OpenAI."""
    from openai import OpenAI
    oai = OpenAI()
    response = oai.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding


# ── Step 3: Build the index ──
def build_index(documents: list[str]) -> dict:
    """Chunk documents and compute embeddings for each chunk."""
    all_chunks   = []
    all_embeddings = []
    for doc in documents:
        chunks = chunk_text(doc)
        for chunk in chunks:
            all_chunks.append(chunk)
            all_embeddings.append(get_embedding(chunk))
            print(f"Indexed chunk {len(all_chunks)}")
    return {"chunks": all_chunks, "embeddings": np.array(all_embeddings)}


# ── Step 4: Semantic search ──
def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def search(query: str, index: dict, top_k: int = 3) -> list[str]:
    """Find the most relevant chunks for a query."""
    query_emb = np.array(get_embedding(query))
    scores = [
        cosine_similarity(query_emb, emb)
        for emb in index["embeddings"]
    ]
    top_indices = np.argsort(scores)[:-top_k-1:-1]
    return [index["chunks"][i] for i in top_indices]


# ── Step 5: RAG query ──
def rag_query(question: str, index: dict) -> str:
    """Retrieve relevant chunks and ask Claude to answer."""
    chunks   = search(question, index, top_k=3)
    context  = "\n\n---\n\n".join(chunks)

    prompt = f"""Answer the question using ONLY the provided context.
If the answer is not in the context, say "I don't have that information."

CONTEXT:
{context}

QUESTION: {question}

ANSWER:"""

    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text


# ── Put it all together ──
docs = [
    Path("docs/readme.md").read_text(),
    Path("docs/api_guide.md").read_text(),
]
index = build_index(docs)
answer = rag_query("How do I authenticate with the API?", index)
print(answer)

import anthropic

client = anthropic.Anthropic()

# ── Count tokens BEFORE sending (saves money during development) ──
response = client.messages.count_tokens(
    model="claude-sonnet-4-5",
    system="You are a helpful assistant.",
    messages=[{"role": "user", "content": "Summarize the history of Python."}]
)
print(f"Input tokens: {response.input_tokens}")

# ── Cost tracker ──
# claude-sonnet-4-5 pricing (check anthropic.com for current rates)
INPUT_COST_PER_1K  = 0.003   # $0.003 per 1K input tokens
OUTPUT_COST_PER_1K = 0.015   # $0.015 per 1K output tokens

def estimate_cost(input_tokens: int, output_tokens: int) -> float:
    return (
        (input_tokens  / 1000) * INPUT_COST_PER_1K +
        (output_tokens / 1000) * OUTPUT_COST_PER_1K
    )

# Track costs across your app
total_cost = 0.0

def tracked_message(**kwargs) -> anthropic.types.Message:
    global total_cost
    response = client.messages.create(**kwargs)
    cost = estimate_cost(
        response.usage.input_tokens,
        response.usage.output_tokens
    )
    total_cost += cost
    print(f"[cost] ${cost:.5f} | total: ${total_cost:.4f}")
    return response

import anthropic, time, logging
from functools import wraps

log = logging.getLogger(__name__)

def with_retry(max_retries: int = 3, base_delay: float = 1.0):
    """Decorator for retrying LLM calls with exponential backoff."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)

                except anthropic.RateLimitError as e:
                    # Rate limited — wait and retry
                    wait = base_delay * (2 ** attempt)
                    log.warning(f"Rate limited. Waiting {wait}s (attempt {attempt+1}/{max_retries})")
                    time.sleep(wait)

                except anthropic.APIStatusError as e:
                    if e.status_code >= 500:
                        # Server error — retry
                        wait = base_delay * (2 ** attempt)
                        log.warning(f"Server error {e.status_code}. Retry in {wait}s")
                        time.sleep(wait)
                    else:
                        # Client error (400, 401, 403) — don't retry
                        log.error(f"Client error {e.status_code}: {e.message}")
                        raise

                except anthropic.APIConnectionError:
                    # Network issue — retry
                    wait = base_delay * (2 ** attempt)
                    log.warning(f"Connection error. Retry in {wait}s")
                    time.sleep(wait)

            raise RuntimeError(f"API call failed after {max_retries} attempts")
        return wrapper
    return decorator


@with_retry(max_retries=3)
def safe_chat(message: str) -> str:
    client = anthropic.Anthropic()
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": message}]
    )
    return response.content[0].text