# Function calling: strict mode tool definition
tools = [{
    "type": "function",
    "function": {
        "name": "search_documents",
        "description": "Search knowledge base",
        "strict": True,
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query"},
                "limit": {"type": "integer", "description": "Max results"}
            },
            "required": ["query", "limit"],
            "additionalProperties": False
        }
    }
}]

# Streaming: SSE endpoint with FastAPI
@app.get("/chat/stream")
async def stream_chat(prompt: str):
    async def generate():
        async for token in async_stream(prompt):
            yield {"event": "token", "data": token}
        yield {"event": "done", "data": ""}
    return EventSourceResponse(generate())

# Local inference: Ollama with LangChain
llm = ChatOllama(
    model="deepseek-r1:70b",
    base_url="http://localhost:11434",
    temperature=0.0,
    num_ctx=32768,
)

# Fine-tuning: QLoRA with Unsloth
model, tokenizer = FastLanguageModel.from_pretrained(
    model_name="unsloth/Meta-Llama-3.1-8B",
    max_seq_length=2048, load_in_4bit=True,
)
model = FastLanguageModel.get_peft_model(model, r=16, lora_alpha=32)

# OpenAI supports parallel tool calls
response = await llm.chat(
    messages=messages,
    tools=tools,
    parallel_tool_calls=True  # Default in GPT-5 series
)

# Handle multiple calls in parallel
if response.tool_calls:
    results = await asyncio.gather(*[
        execute_tool(tc.function.name, json.loads(tc.function.arguments))
        for tc in response.tool_calls
    ])

# Structured outputs with strict=True may not work with parallel_tool_calls
# If using strict mode schemas, disable parallel calls:
response = await llm.chat(
    messages=messages,
    tools=tools_with_strict_true,
    parallel_tool_calls=False  # Required for strict mode reliability
)

async def execute_tools_parallel(tool_calls: list) -> list[dict]:
    """Execute tool calls in parallel with error handling."""
    async def safe_execute(tc):
        try:
            result = await execute_tool(
                tc.function.name,
                json.loads(tc.function.arguments)
            )
            return {"tool_call_id": tc.id, "content": json.dumps(result)}
        except Exception as e:
            return {"tool_call_id": tc.id, "content": json.dumps({"error": str(e)})}

    results = await asyncio.gather(*[safe_execute(tc) for tc in tool_calls])
    return [{"role": "tool", **r} for r in results]

# Crashes entire batch if one tool fails
response = await llm.chat(messages=messages, tools=tools, parallel_tool_calls=True)
results = await asyncio.gather(*[
    execute_tool(tc.function.name, json.loads(tc.function.arguments))
    for tc in response.tool_calls
])

async def safe_execute(tc):
    try:
        result = await execute_tool(tc.function.name, json.loads(tc.function.arguments))
        return {"tool_call_id": tc.id, "content": json.dumps(result)}
    except Exception as e:
        return {"tool_call_id": tc.id, "content": json.dumps({"error": str(e)})}

results = await asyncio.gather(*[safe_execute(tc) for tc in response.tool_calls])

# OpenAI format with strict mode (2026 recommended)
tools = [{
    "type": "function",
    "function": {
        "name": "search_documents",
        "description": "Search the document database for relevant content",
        "strict": True,  # Enables structured output validation
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The search query"
                },
                "limit": {
                    "type": "integer",
                    "description": "Max results to return"
                }
            },
            "required": ["query", "limit"],  # All props required when strict
            "additionalProperties": False     # Required for strict mode
        }
    }
}]

# Note: With strict=True:
# - All properties must be listed in "required"
# - additionalProperties must be False
# - No "default" values (provide via code instead)

def create_tool_schema(
    name: str,
    description: str,
    parameters: dict,
    strict: bool = True
) -> dict:
    """Create OpenAI-compatible tool schema with strict mode."""
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": description,
            "strict": strict,
            "parameters": {
                "type": "object",
                "properties": parameters,
                "required": list(parameters.keys()),
                "additionalProperties": False
            }
        }
    }

def create_anthropic_tool(
    name: str,
    description: str,
    input_schema: dict
) -> dict:
    """Create Anthropic-compatible tool definition."""
    return {
        "name": name,
        "description": description,
        "input_schema": {
            "type": "object",
            "properties": input_schema,
            "required": list(input_schema.keys())
        }
    }

from langchain_core.tools import tool
from pydantic import BaseModel, Field

@tool
def search_documents(query: str, limit: int = 5) -> list[dict]:
    """Search the document database.

    Args:
        query: Search query string
        limit: Maximum results to return
    """
    return db.search(query, limit=limit)

# Bind to model
llm_with_tools = llm.bind_tools([search_documents])

# Or with structured output
class SearchResult(BaseModel):
    query: str = Field(description="The search query used")
    results: list[str] = Field(description="Matching documents")

structured_llm = llm.with_structured_output(SearchResult)

from pydantic import BaseModel

class Analysis(BaseModel):
    sentiment: str
    confidence: float
    key_points: list[str]

# OpenAI structured output
response = await client.beta.chat.completions.parse(
    model="gpt-5.2",
    messages=[{"role": "user", "content": "Analyze this text..."}],
    response_format=Analysis
)

analysis = response.choices[0].message.parsed  # Typed Analysis object

tools = [{
    "type": "function",
    "function": {
        "name": "search",
        "strict": True,
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "limit": {"type": "integer", "default": 10}  # Invalid with strict
            },
            "required": ["query"]  # Must include all props when strict=True
        }
    }
}]

tools = [{
    "type": "function",
    "function": {
        "name": "search",
        "strict": True,
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "limit": {"type": "integer"}
            },
            "required": ["query", "limit"],  # All properties required
            "additionalProperties": False    # Required for strict mode
        }
    }
}]

async def run_with_tools(messages: list, tools: list) -> str:
    """Execute tool calls until LLM returns final answer."""
    while True:
        response = await llm.chat(messages=messages, tools=tools)

        # Check if LLM wants to call tools
        if not response.tool_calls:
            return response.content

        # Execute each tool call
        for tool_call in response.tool_calls:
            result = await execute_tool(
                tool_call.function.name,
                json.loads(tool_call.function.arguments)
            )

            # Add tool result to conversation
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result)
            })

        # Continue loop (LLM will process tool results)

class ToolRegistry:
    """Registry for managing tool definitions and execution."""

    def __init__(self):
        self.tools: dict[str, Callable] = {}
        self.schemas: list[dict] = []

    def register(self, func: Callable) -> Callable:
        """Register a function as a tool."""
        schema = self._extract_schema(func)
        self.tools[func.__name__] = func
        self.schemas.append(schema)
        return func

    async def execute(self, name: str, args: dict) -> Any:
        """Execute a registered tool with validation."""
        if name not in self.tools:
            raise ValueError(f"Unknown tool: {name}")
        func = self.tools[name]
        if asyncio.iscoroutinefunction(func):
            return await func(**args)
        return func(**args)

async def run_tool_loop(
    registry: ToolRegistry,
    user_message: str,
    model: str = "gpt-5.2",
    max_iterations: int = 10
) -> str:
    """Run tool execution loop with iteration guard."""
    client = AsyncOpenAI()
    messages = [{"role": "user", "content": user_message}]

    for _ in range(max_iterations):
        response = await client.chat.completions.create(
            model=model,
            messages=messages,
            tools=registry.schemas,
            parallel_tool_calls=False
        )

        message = response.choices[0].message
        if not message.tool_calls:
            return message.content

        messages.append(message.model_dump())

        for tool_call in message.tool_calls:
            try:
                result = await registry.execute(
                    tool_call.function.name,
                    json.loads(tool_call.function.arguments)
                )
            except Exception as e:
                result = {"error": str(e)}

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result)
            })

    raise RuntimeError("Max iterations reached")

async def run_tools(user_message: str):
    messages = [{"role": "user", "content": user_message}]
    while True:  # Infinite loop risk
        response = await llm.chat(messages=messages, tools=tools)
        if not response.tool_calls:
            return response.content
        # Execute tools and continue...

async def run_tools(user_message: str, max_iterations: int = 10):
    messages = [{"role": "user", "content": user_message}]
    for _ in range(max_iterations):
        response = await llm.chat(messages=messages, tools=tools)
        if not response.tool_calls:
            return response.content
        # Execute tools and continue...
    raise RuntimeError("Max iterations reached")

# Loading entire knowledge base into every request
context = load_all_documents() + load_all_examples()
response = llm.chat(system=context, messages=[user_msg])
# Wastes tokens, hits limits, degrades quality

# Just-in-time document loading with token budget
async def build_context(query: str, budget: int) -> list[dict]:
    # Retrieve only relevant documents
    relevant_docs = await retriever.search(query, top_k=5)

    # Truncate each doc to fit budget
    doc_budget = int(budget * 0.25)  # 25% for retrieval
    truncated = [truncate_to_tokens(doc, doc_budget // len(relevant_docs))
                 for doc in relevant_docs]

    return truncated

# Validate compression quality with functional probes
PROBES = [
    "What is the session intent?",
    "What files were modified?",
    "What decisions were made and why?",
]

async def evaluate_compression(summary: str) -> float:
    passed = 0
    for probe in PROBES:
        response = await llm.answer(f"Based on this summary:\n{summary}\n\n{probe}")
        if response_is_valid(response):
            passed += 1
    return passed / len(PROBES)  # Target: >90% pass rate

# Stuffing entire conversation into context without structure
messages = full_history + retrieved_docs + system_prompt
response = llm.chat(messages)  # Hits limits, "lost in the middle" recall drops to 10-40%

# Five-layer context architecture with attention-aware positioning
ALLOCATIONS = {
    "agent": {
        "system": 0.10,       # 10% — START (high attention)
        "tools": 0.15,        # 15% — START
        "history": 0.30,      # 30% — MIDDLE (compressible)
        "retrieval": 0.25,    # 25% — MIDDLE (just-in-time)
        "observations": 0.20, # 20% — END (high attention)
    },
}

# Compression triggers
COMPRESS_AT = 0.70   # 70% utilization
TARGET_AFTER = 0.50  # 50% utilization after compression
MIN_MESSAGES = 10    # Minimum before compressing
PRESERVE_LAST = 5    # Always keep last 5 uncompressed

## Session Intent
[What we're trying to accomplish - NEVER lose this]

## Files Modified
- path/to/file.ts: Added function X, modified class Y

## Decisions Made
- Decision 1: Chose X over Y because [rationale]

## Current State
[Where we are in the task - progress indicator]

## Next Steps
1. Complete X
2. Test Y

# Returning raw LLM output without validation
response = await llm.generate(prompt)
return response  # No quality check!

QUALITY_THRESHOLD = 0.7

async def quality_gate(state: dict) -> dict:
    """Gate LLM output with multi-metric assessment."""
    scores = await full_quality_assessment(state["input"], state["output"])
    passed = scores["average"] >= QUALITY_THRESHOLD
    return {
        **state,
        "quality_passed": passed,
        "scores": scores,
        "retry_count": state.get("retry_count", 0) + (0 if passed else 1),
    }

async def full_quality_assessment(input_text: str, output_text: str) -> dict:
    dimensions = ["relevance", "accuracy", "completeness"]
    scores = {}
    for dim in dimensions:
        scores[dim] = await evaluate_quality(input_text, output_text, dim)
    scores["average"] = sum(scores.values()) / len(scores)
    return scores

async def batch_evaluate(model, dataset: list[dict], metrics: list[str]) -> dict:
    """Evaluate model over a golden dataset."""
    results = []
    for example in dataset:
        output = await model.generate(example["input"])
        scores = {m: await evaluate(example, output, m) for m in metrics}
        results.append({"input": example["input"], "expected": example["expected"],
                        "actual": output, "scores": scores})

    # Aggregate
    avg_scores = {m: sum(r["scores"][m] for r in results) / len(results) for m in metrics}
    return {"sample_size": len(dataset), "avg_scores": avg_scores, "results": results}

async def pairwise_compare(input_text: str, output_a: str, output_b: str) -> str:
    """Compare two model outputs, return winner."""
    response = await judge_model.chat([{
        "role": "user",
        "content": f"""Compare these two responses to the input.
Input: {input_text[:500]}
Response A: {output_a[:1000]}
Response B: {output_b[:1000]}
Which is better? Reply with just 'A' or 'B'."""
    }])
    return response.content.strip()

# Only checking one thing with same model as judge
output = await gpt4.complete(prompt)
score = await gpt4.evaluate(output)  # Same model as judge!
if score > 0.95:  # Threshold too high, blocks most content
    return "pass"

async def evaluate_quality(input_text: str, output_text: str, dimension: str) -> float:
    """Use a DIFFERENT model as judge."""
    response = await judge_model.chat([{
        "role": "user",
        "content": f"""Evaluate for {dimension}. Score 1-10.
Input: {input_text[:500]}
Output: {output_text[:1000]}
Respond with just the number."""
    }])
    return int(response.content.strip()) / 10

# Evaluate across 3-5 dimensions
dimensions = ["relevance", "accuracy", "completeness", "coherence"]
scores = {d: await evaluate_quality(input_text, output, d) for d in dimensions}
average = sum(scores.values()) / len(scores)
passed = average >= 0.7  # 0.7 for production, 0.6 for drafts

async def detect_hallucination(context: str, output: str) -> dict:
    """Check if output contains claims not supported by context."""
    response = await judge_model.chat([{
        "role": "user",
        "content": f"""Check if the output contains claims not in the context.
Context: {context[:2000]}
Output: {output[:1000]}
List any unsupported claims."""
    }])
    return {"has_hallucinations": bool(unsupported), "unsupported_claims": unsupported}

import os
from langchain_ollama import ChatOllama

def get_llm_provider(task_type: str = "general"):
    """Auto-switch between Ollama and cloud APIs."""
    if os.getenv("OLLAMA_ENABLED") == "true":
        models = {
            "reasoning": "deepseek-r1:70b",
            "coding": "qwen2.5-coder:32b",
            "general": "llama3.3:70b",
        }
        return ChatOllama(
            model=models.get(task_type, "llama3.3:70b"),
            keep_alive="5m"
        )
    else:
        # Fall back to cloud API
        from langchain_openai import ChatOpenAI
        return ChatOpenAI(model="gpt-5.2")

# Usage
llm = get_llm_provider(task_type="coding")

from pydantic import BaseModel, Field

class CodeAnalysis(BaseModel):
    language: str = Field(description="Programming language")
    complexity: int = Field(ge=1, le=10)
    issues: list[str] = Field(description="Found issues")

structured_llm = llm.with_structured_output(CodeAnalysis)
result = await structured_llm.ainvoke("Analyze this code: ...")
# result is typed CodeAnalysis object

# GitHub Actions (self-hosted runner)
jobs:
  test:
    runs-on: self-hosted  # M4 Max runner
    env:
      OLLAMA_ENABLED: "true"
    steps:
      - name: Pre-warm models
        run: |
          curl -s http://localhost:11434/api/embeddings \
            -d '{"model":"nomic-embed-text","prompt":"warmup"}' > /dev/null

      - name: Run tests
        run: pytest tests/

import httpx

async def prewarm_models() -> None:
    """Pre-warm Ollama models for faster first request."""
    async with httpx.AsyncClient() as client:
        # Warm embedding model
        await client.post(
            "http://localhost:11434/api/embeddings",
            json={"model": "nomic-embed-text", "prompt": "warmup"},
            timeout=60.0,
        )

        # Warm reasoning model (minimal generation)
        await client.post(
            "http://localhost:11434/api/chat",
            json={
                "model": "deepseek-r1:70b",
                "messages": [{"role": "user", "content": "Hi"}],
                "options": {"num_predict": 1},
            },
            timeout=120.0,
        )

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-5.2")  # Always uses cloud, ignores local setup
response = await llm.ainvoke("Generate code...")

import os
from langchain_ollama import ChatOllama
from langchain_openai import ChatOpenAI

def get_llm_provider(task_type: str = "general"):
    if os.getenv("OLLAMA_ENABLED") == "true":
        return ChatOllama(model="qwen2.5-coder:32b", keep_alive="5m")
    return ChatOpenAI(model="gpt-5.2")

llm = get_llm_provider(task_type="coding")

HARDWARE_PROFILES = {
    "m4_max_256gb": {
        "reasoning": "deepseek-r1:70b",
        "coding": "qwen2.5-coder:32b",
        "general": "llama3.3:70b",
        "embeddings": "nomic-embed-text",
        "max_loaded": 3
    },
    "m3_pro_36gb": {
        "reasoning": "llama3.3:7b",
        "coding": "qwen2.5-coder:7b",
        "general": "llama3.3:7b",
        "embeddings": "nomic-embed-text",
        "max_loaded": 2
    },
    "ci_runner": {
        "all": "llama3.3:7b",  # Fast, low memory
        "embeddings": "nomic-embed-text",
        "max_loaded": 1
    }
}

def get_model_for_task(task: str, hardware: str = "m4_max_256gb") -> str:
    """Select model based on task and available hardware."""
    profile = HARDWARE_PROFILES[hardware]
    return profile.get(task, profile.get("general", "llama3.3:7b"))

# Full precision (best quality, most VRAM)
ollama pull deepseek-r1:70b

# Q4_K_M quantization (good balance)
ollama pull deepseek-r1:70b-q4_K_M

# Q4_0 quantization (fastest, lowest quality)
ollama pull deepseek-r1:70b-q4_0

# M3 Pro 36GB trying to run 70B model
llm = ChatOllama(model="deepseek-r1:70b")  # OOM error, 42GB VRAM needed
response = await llm.ainvoke("Simple task")

def get_model_for_hardware(hardware: str, task: str) -> str:
    profiles = {
        "m3_pro_36gb": {"reasoning": "llama3.3:7b"},
        "m4_max_256gb": {"reasoning": "deepseek-r1:70b"}
    }
    return profiles[hardware].get(task, "llama3.3:7b")

model = get_model_for_hardware("m3_pro_36gb", "reasoning")
llm = ChatOllama(model=model)

# Install Ollama
curl -fsSL https://ollama.ai/install.sh | sh

# Pull models
ollama pull deepseek-r1:70b      # Reasoning (GPT-4 level)
ollama pull qwen2.5-coder:32b    # Coding
ollama pull nomic-embed-text     # Embeddings

# Start server
ollama serve

from langchain_ollama import ChatOllama, OllamaEmbeddings

# Chat model
llm = ChatOllama(
    model="deepseek-r1:70b",
    base_url="http://localhost:11434",
    temperature=0.0,
    num_ctx=32768,      # Context window
    keep_alive="5m",    # Keep model loaded
)

# Embeddings
embeddings = OllamaEmbeddings(
    model="nomic-embed-text",
    base_url="http://localhost:11434",
)

# Generate
response = await llm.ainvoke("Explain async/await")
vector = await embeddings.aembed_query("search text")

from langchain_core.tools import tool

@tool
def search_docs(query: str) -> str:
    """Search the document database."""
    return f"Found results for: {query}"

# Bind tools
llm_with_tools = llm.bind_tools([search_docs])
response = await llm_with_tools.ainvoke("Search for Python patterns")

# .env.local
OLLAMA_ENABLED=true
OLLAMA_HOST=http://localhost:11434
OLLAMA_MODEL_REASONING=deepseek-r1:70b
OLLAMA_MODEL_CODING=qwen2.5-coder:32b
OLLAMA_MODEL_EMBED=nomic-embed-text

# Performance tuning (Apple Silicon)
OLLAMA_MAX_LOADED_MODELS=3    # Keep 3 models in memory
OLLAMA_KEEP_ALIVE=5m          # 5 minute keep-alive

# Check if Ollama is running
curl http://localhost:11434/api/tags

# List loaded models
ollama list

# Check model memory usage
ollama ps

# Pull specific quantization
ollama pull deepseek-r1:70b-q4_K_M

from langchain_ollama import ChatOllama

llm = ChatOllama(model="deepseek-r1:70b")  # Model unloaded after each call
response = await llm.ainvoke("Task 1")  # 30-60s cold start
response = await llm.ainvoke("Task 2")  # Another 30-60s cold start

from langchain_ollama import ChatOllama

llm = ChatOllama(
    model="deepseek-r1:70b",
    keep_alive="5m"  # Keep model loaded for 5 minutes
)
response = await llm.ainvoke("Task 1")  # 30-60s initial load
response = await llm.ainvoke("Task 2")  # Instant (model still loaded)

# No reasoning structure for complex problems
response = llm.complete("Solve: 15% of 240")  # No CoT!

# Single example for few-shot (too few)
examples = [{"input": "x", "output": "y"}]

# Hardcoded prompt without versioning
PROMPT = "You are a helpful assistant..."  # No version control!

COT_SYSTEM = """You are a helpful assistant that solves problems step-by-step.

When solving problems:
1. Break down the problem into clear steps
2. Show your reasoning for each step
3. Verify your answer before responding
4. If uncertain, acknowledge limitations

Format your response as:
STEP 1: [description]
Reasoning: [your thought process]
FINAL ANSWER: [your conclusion]"""

cot_prompt = ChatPromptTemplate.from_messages([
    ("system", COT_SYSTEM),
    ("human", "Problem: {problem}\n\nThink through this step-by-step."),
])

from langchain_core.prompts import FewShotChatMessagePromptTemplate

# Use 3-5 diverse, representative examples
examples = [ex1, ex2, ex3, ex4, ex5]

few_shot = FewShotChatMessagePromptTemplate(
    examples=examples,
    example_prompt=ChatPromptTemplate.from_messages([
        ("human", "{input}"),
        ("ai", "{output}"),
    ]),
)

# Most similar examples last (recency bias helps)
final_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant. Answer concisely."),
    few_shot,
    ("human", "{input}"),
])

# Hardcoded prompt, no version control, no A/B testing
PROMPT = "You are a helpful assistant..."
response = llm.complete(PROMPT + user_input)
# No way to know if prompt changes improve or degrade quality

from langfuse import Langfuse

langfuse = Langfuse()

# Get versioned prompt with environment label
prompt = langfuse.get_prompt(
    name="customer-support-v2",
    label="production",  # production, staging, canary
    cache_ttl_seconds=300,
)

# Compile with variables
compiled = prompt.compile(
    customer_name="John",
    issue="billing question"
)

# Track via trace metadata for A/B comparison
trace = langfuse.trace(
    name="support-query",
    metadata={"prompt_version": prompt.version, "variant": "A"},
)

import dspy

class OptimizedQA(dspy.Module):
    def __init__(self):
        self.generate = dspy.Predict("question -> answer")

    def forward(self, question):
        return self.generate(question=question)

# MIPROv2: Data+demo-aware Bayesian optimization (recommended)
optimizer = dspy.MIPROv2(metric=answer_match)
optimized = optimizer.compile(OptimizedQA(), trainset=examples)

# Alternative: GEPA (July 2025) - Reflective Prompt Evolution
# Uses model introspection to analyze failures and propose better prompts

async def self_consistent_answer(question: str, n_paths: int = 5) -> str:
    """Generate multiple CoT reasoning paths and vote on answer."""
    answers = []
    for _ in range(n_paths):
        response = await llm.chat([{
            "role": "user",
            "content": f"{question}\n\nThink step by step."
        }], temperature=0.7)  # Higher temp for diversity
        answer = extract_final_answer(response)
        answers.append(answer)

    # Majority vote
    from collections import Counter
    return Counter(answers).most_common(1)[0][0]

import asyncio

async def stream_with_backpressure(prompt: str, max_buffer: int = 100):
    """Handle slow consumers with backpressure."""
    buffer = asyncio.Queue(maxsize=max_buffer)

    async def producer():
        async for token in async_stream(prompt):
            await buffer.put(token)  # Blocks if buffer full
        await buffer.put(None)  # Signal completion

    async def consumer():
        while True:
            token = await buffer.get()
            if token is None:
                break
            yield token
            await asyncio.sleep(0)  # Yield control

    # Start producer in background
    asyncio.create_task(producer())

    # Return consumer generator
    async for token in consumer():
        yield token

// Frontend: Cancel with AbortController
const controller = new AbortController();

async function streamChat(prompt: string, onToken: (t: string) => void) {
  const response = await fetch("/chat/stream?prompt=" + encodeURIComponent(prompt), {
    signal: controller.signal
  });

  const reader = response.body?.getReader();
  const decoder = new TextDecoder();

  try {
    while (reader) {
      const { done, value } = await reader.read();
      if (done) break;
      onToken(decoder.decode(value));
    }
  } catch (err) {
    if (err.name === 'AbortError') {
      console.log('Stream cancelled by user');
    }
  }
}

// Cancel the stream
controller.abort();

from fastapi import Request

@app.get("/chat/stream")
async def stream_chat(prompt: str, request: Request):
    """SSE with server-side disconnect detection."""
    async def generate():
        async for token in async_stream(prompt):
            if await request.is_disconnected():
                break  # Client disconnected
            yield {"event": "token", "data": token}
        yield {"event": "done", "data": ""}

    return EventSourceResponse(generate())

async def stream_tokens(prompt: str):
    buffer = asyncio.Queue()  # No maxsize = unbounded
    async for token in async_stream(prompt):
        await buffer.put(token)  # Never blocks, grows infinitely
    # Slow consumer = OOM

async def stream_tokens(prompt: str):
    buffer = asyncio.Queue(maxsize=100)  # Bounded buffer
    async for token in async_stream(prompt):
        await buffer.put(token)  # Blocks when full, slows producer
    # Producer matches consumer speed

from openai import AsyncOpenAI

client = AsyncOpenAI()

async def async_stream(prompt: str):
    """Async streaming for better concurrency."""
    stream = await client.chat.completions.create(
        model="gpt-5.2",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )

    async for chunk in stream:
        if chunk.choices[0].delta.content:
            yield chunk.choices[0].delta.content

from fastapi import FastAPI
from sse_starlette.sse import EventSourceResponse

app = FastAPI()

@app.get("/chat/stream")
async def stream_chat(prompt: str):
    """Server-Sent Events endpoint for streaming."""
    async def generate():
        async for token in async_stream(prompt):
            yield {
                "event": "token",
                "data": token
            }
        yield {"event": "done", "data": ""}

    return EventSourceResponse(generate())

async function streamChat(prompt: string, onToken: (t: string) => void) {
  const response = await fetch("/chat/stream?prompt=" + encodeURIComponent(prompt));
  const reader = response.body?.getReader();
  const decoder = new TextDecoder();

  while (reader) {
    const { done, value } = await reader.read();
    if (done) break;

    const text = decoder.decode(value);
    const lines = text.split('\n');

    for (const line of lines) {
      if (line.startsWith('data: ')) {
        const data = line.slice(6);
        if (data !== '[DONE]') {
          onToken(data);
        }
      }
    }
  }
}

// Usage
let fullResponse = '';
await streamChat('Hello', (token) => {
  fullResponse += token;
  setDisplayText(fullResponse);  // Update UI incrementally
});

@app.get("/chat/stream")
async def stream_chat(prompt: str):
    full_response = ""
    async for token in async_stream(prompt):
        full_response += token  # Accumulate everything
    return {"response": full_response}  # Send all at once

@app.get("/chat/stream")
async def stream_chat(prompt: str):
    async def generate():
        async for token in async_stream(prompt):
            yield {"event": "token", "data": token}  # Send immediately
    return EventSourceResponse(generate())

async def stream_with_tools(messages: list, tools: list):
    """Handle streaming responses that include tool calls."""
    stream = await client.chat.completions.create(
        model="gpt-5.2",
        messages=messages,
        tools=tools,
        stream=True
    )

    collected_content = ""
    collected_tool_calls = []

    async for chunk in stream:
        delta = chunk.choices[0].delta

        # Collect content tokens
        if delta.content:
            collected_content += delta.content
            yield {"type": "content", "data": delta.content}

        # Collect tool call chunks
        if delta.tool_calls:
            for tc in delta.tool_calls:
                # Tool calls come in chunks, accumulate them
                if tc.index >= len(collected_tool_calls):
                    collected_tool_calls.append({
                        "id": tc.id,
                        "function": {"name": "", "arguments": ""}
                    })

                if tc.function.name:
                    collected_tool_calls[tc.index]["function"]["name"] += tc.function.name
                if tc.function.arguments:
                    collected_tool_calls[tc.index]["function"]["arguments"] += tc.function.arguments

    # If tool calls, execute them
    if collected_tool_calls:
        yield {"type": "tool_calls", "data": collected_tool_calls}

import json

def try_parse_partial_json(buffer: str) -> dict | None:
    """Attempt to parse partial JSON, returning None if incomplete."""
    try:
        return json.loads(buffer)
    except json.JSONDecodeError:
        return None

async def stream_structured_output(prompt: str):
    """Stream and incrementally parse structured output."""
    buffer = ""
    async for token in async_stream(prompt):
        buffer += token
        parsed = try_parse_partial_json(buffer)
        if parsed:
            yield {"type": "parsed", "data": parsed}
        else:
            yield {"type": "partial", "data": buffer}

async for chunk in stream:
    if chunk.choices[0].delta.tool_calls:
        tc = chunk.choices[0].delta.tool_calls[0]
        # Parse before accumulation completes
        args = json.loads(tc.function.arguments)  # JSONDecodeError on partial data
        execute_tool(tc.function.name, args)

collected_tool_calls = []
async for chunk in stream:
    if chunk.choices[0].delta.tool_calls:
        for tc in chunk.choices[0].delta.tool_calls:
            if tc.index >= len(collected_tool_calls):
                collected_tool_calls.append({"function": {"arguments": ""}})
            collected_tool_calls[tc.index]["function"]["arguments"] += tc.function.arguments

# Parse after stream completes
for tc in collected_tool_calls:
    args = json.loads(tc["function"]["arguments"])

import json
import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def generate_training_example(topic: str) -> dict:
    """Generate a single training example using teacher model."""
    response = await client.chat.completions.create(
        model="gpt-5.2",  # Teacher
        messages=[{
            "role": "system",
            "content": f"Generate a training example about {topic}. "
                      "Include instruction and response."
        }],
        response_format={"type": "json_object"},
        temperature=0.9,  # Higher for diversity
    )
    return json.loads(response.choices[0].message.content)


async def generate_dataset(topic: str, num_examples: int = 100) -> list[dict]:
    """Generate dataset in batches."""
    examples = []
    batch_size = 10

    for batch_start in range(0, num_examples, batch_size):
        batch_tasks = [
            generate_training_example(topic)
            for _ in range(min(batch_size, num_examples - batch_start))
        ]
        batch_results = await asyncio.gather(*batch_tasks)
        examples.extend(batch_results)

    return examples

async def validate_example(example: dict, validator_model: str = "gpt-5.2-mini") -> dict:
    """Validate and score a training example."""
    response = await client.chat.completions.create(
        model=validator_model,
        messages=[{
            "role": "system",
            "content": """Score this training example 1-10 on:
- clarity: Is the instruction clear?
- quality: Is the response high quality?
- realism: Is this a realistic interaction?

Output JSON: {"clarity": N, "quality": N, "realism": N, "keep": true/false}
Set keep=false if any score < 6."""
        }, {
            "role": "user",
            "content": f"Instruction: {example['instruction']}\n\nResponse: {example['response']}"
        }],
        response_format={"type": "json_object"},
    )
    return {**example, **json.loads(response.choices[0].message.content)}

from sentence_transformers import SentenceTransformer
import numpy as np

def deduplicate_examples(examples: list[dict], threshold: float = 0.85) -> list[dict]:
    """Remove near-duplicate examples using embeddings."""
    model = SentenceTransformer("all-MiniLM-L6-v2")
    instructions = [ex["instruction"] for ex in examples]
    embeddings = model.encode(instructions)

    unique_indices = []
    for i, emb in enumerate(embeddings):
        is_unique = True
        for j in unique_indices:
            similarity = np.dot(emb, embeddings[j]) / (
                np.linalg.norm(emb) * np.linalg.norm(embeddings[j])
            )
            if similarity > threshold:
                is_unique = False
                break
        if is_unique:
            unique_indices.append(i)

    return [examples[i] for i in unique_indices]

# Alpaca format
def to_alpaca_format(examples: list[dict]) -> list[dict]:
    return [{
        "instruction": ex["instruction"],
        "input": ex.get("input", ""),
        "output": ex["response"],
    } for ex in examples]

# ChatML format
def to_chatml_format(examples: list[dict]) -> list[dict]:
    return [{
        "messages": [
            {"role": "user", "content": ex["instruction"]},
            {"role": "assistant", "content": ex["response"]},
        ]
    } for ex in examples]

async def generate_dataset(topic: str, num: int = 1000):
    examples = []
    for _ in range(num):
        ex = await generate_example(topic)
        examples.append(ex)  # No validation, possible duplicates
    return examples

async def generate_dataset(topic: str, num: int = 1000):
    examples = []
    for _ in range(num):
        ex = await generate_example(topic)
        validation = await validate_example(ex)
        if validation["keep"]:  # Filter low-quality
            examples.append(ex)
    return deduplicate_examples(examples, threshold=0.85)

from trl import DPOTrainer, DPOConfig

config = DPOConfig(
    learning_rate=5e-6,  # Lower for alignment
    beta=0.1,            # KL penalty coefficient
    per_device_train_batch_size=4,
    num_train_epochs=1,
)

# Preference dataset: {prompt, chosen, rejected}
trainer = DPOTrainer(
    model=model,
    ref_model=ref_model,  # Frozen reference
    args=config,
    train_dataset=preference_dataset,
    tokenizer=tokenizer,
)
trainer.train()

from peft import LoraConfig, get_peft_model

peft_config = LoraConfig(
    r=16, lora_alpha=32,
    target_modules=["q_proj", "v_proj", "k_proj", "o_proj"],
    lora_dropout=0.05, bias="none", task_type="CAUSAL_LM",
)
model = get_peft_model(model, peft_config)

# With LoRA, no separate ref_model needed
trainer = DPOTrainer(
    model=model,
    ref_model=None,  # Uses implicit reference
    args=DPOConfig(learning_rate=5e-5, beta=0.1),
    train_dataset=dataset,
    tokenizer=tokenizer,
)

async def evaluate_alignment(
    model, tokenizer,
    test_prompts: list[str],
    judge_model: str = "gpt-5.2-mini",
) -> dict:
    """Evaluate model alignment quality."""
    scores = []
    for prompt in test_prompts:
        inputs = tokenizer(prompt, return_tensors="pt")
        outputs = model.generate(**inputs, max_new_tokens=256)
        response = tokenizer.decode(outputs[0], skip_special_tokens=True)

        judgment = await client.chat.completions.create(
            model=judge_model,
            messages=[{
                "role": "user",
                "content": f"Rate this response 1-10 for helpfulness.\n"
                          f"Prompt: {prompt}\nResponse: {response}"
            }]
        )
        scores.append(int(judgment.choices[0].message.content.strip()))

    return {"mean_score": sum(scores) / len(scores), "scores": scores}

# NEVER fine-tune without trying alternatives first
model.fine_tune(data)  # Try prompt engineering & RAG first!

# NEVER use low-quality training data
data = scrape_random_web()  # Garbage in, garbage out

# NEVER skip evaluation
trainer.train()
deploy(model)  # Always evaluate before deploy!

# ALWAYS use separate eval set
train, eval = split(data, test_size=0.1)
trainer = SFTTrainer(..., eval_dataset=eval)

trainer = SFTTrainer(model=model, train_dataset=train_data)
trainer.train()
model.save_pretrained("./production_model")  # No evaluation
deploy(model)  # Could be degraded

train, eval = train_test_split(data, test_size=0.1)
trainer = SFTTrainer(
    model=model,
    train_dataset=train,
    eval_dataset=eval  # Separate eval set
)
trainer.train()
eval_results = await evaluate_alignment(model, tokenizer, test_prompts)
if eval_results["mean_score"] >= 7.5:  # Quality threshold
    deploy(model)

Original: W (4096 x 4096) = 16M parameters
LoRA:     A (4096 x 16) + B (16 x 4096) = 131K parameters (0.8%)

from unsloth import FastLanguageModel
from trl import SFTTrainer

# Load with 4-bit quantization (QLoRA)
model, tokenizer = FastLanguageModel.from_pretrained(
    model_name="unsloth/Meta-Llama-3.1-8B",
    max_seq_length=2048,
    load_in_4bit=True,
)

# Add LoRA adapters
model = FastLanguageModel.get_peft_model(
    model,
    r=16,              # Rank (16-64 typical)
    lora_alpha=32,     # Scaling (2x r)
    lora_dropout=0.05,
    target_modules=[
        "q_proj", "k_proj", "v_proj", "o_proj",  # Attention
        "gate_proj", "up_proj", "down_proj",      # MLP
    ],
)

# Train
trainer = SFTTrainer(
    model=model,
    tokenizer=tokenizer,
    train_dataset=dataset,
    max_seq_length=2048,
)
trainer.train()

from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from transformers import AutoModelForCausalLM, BitsAndBytesConfig
import torch

# 4-bit quantization
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.bfloat16,
    bnb_4bit_use_double_quant=True,
)

model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-3.1-8B",
    quantization_config=bnb_config,
    device_map="auto",
)
model = prepare_model_for_kbit_training(model)

lora_config = LoraConfig(
    r=16, lora_alpha=32,
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj",
                    "gate_proj", "up_proj", "down_proj"],
    lora_dropout=0.05, bias="none", task_type="CAUSAL_LM",
)
model = get_peft_model(model, lora_config)

from peft import PeftModel

base_model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-3.1-8B", torch_dtype=torch.float16, device_map="auto",
)
model = PeftModel.from_pretrained(base_model, "./lora_adapter")
merged_model = model.merge_and_unload()
merged_model.save_pretrained("./merged_model")

model = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-3.1-70B")
trainer = SFTTrainer(model=model, train_dataset=dataset)
trainer.train()  # OOM: requires 560GB+ VRAM

model, tokenizer = FastLanguageModel.from_pretrained(
    "unsloth/Meta-Llama-3.1-70B",
    load_in_4bit=True  # QLoRA quantization
)
model = FastLanguageModel.get_peft_model(model, r=16, lora_alpha=32)
trainer = SFTTrainer(model=model, train_dataset=dataset)
trainer.train()  # Fits in 48GB VRAM

# Each example has: prompt, chosen (good), rejected (bad)
preference_data = [
    {
        "prompt": "Explain quantum computing simply.",
        "chosen": "Quantum computers use qubits that can be both 0 and 1 simultaneously, "
                  "unlike classical bits. This allows them to solve certain problems faster.",
        "rejected": "Quantum computing is very complicated and uses physics stuff. "
                    "It's basically magic computers that are super fast."
    },
    {
        "prompt": "Write a professional email declining a meeting.",
        "chosen": "Subject: Re: Meeting Request\n\nThank you for the invitation. "
                  "Unfortunately, I have a prior commitment at that time. "
                  "Could we reschedule to later this week?",
        "rejected": "Can't make it, too busy. Maybe some other time idk."
    }
]

from trl import DPOTrainer, DPOConfig
from transformers import AutoModelForCausalLM, AutoTokenizer
from datasets import Dataset

# Load SFT'd model (DPO requires supervised fine-tuned base)
model = AutoModelForCausalLM.from_pretrained(
    "your-sft-model",
    torch_dtype=torch.float16,
    device_map="auto",
)
tokenizer = AutoTokenizer.from_pretrained("your-sft-model")
tokenizer.pad_token = tokenizer.eos_token

# Reference model (frozen copy for KL constraint)
ref_model = AutoModelForCausalLM.from_pretrained(
    "your-sft-model",
    torch_dtype=torch.float16,
    device_map="auto",
)

# DPO configuration
config = DPOConfig(
    # Learning rate (lower than SFT)
    learning_rate=5e-7,

    # Beta: KL penalty coefficient
    # Higher = closer to reference, Lower = more aggressive alignment
    beta=0.1,

    # Sequence lengths
    max_length=1024,
    max_prompt_length=512,

    # Training
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,
    num_train_epochs=1,

    # Optimization
    warmup_ratio=0.1,
    weight_decay=0.01,

    # Logging
    logging_steps=10,
    output_dir="./dpo_output",
)

# Prepare dataset
dataset = Dataset.from_list(preference_data)

# Create trainer
trainer = DPOTrainer(
    model=model,
    ref_model=ref_model,
    args=config,
    train_dataset=dataset,
    tokenizer=tokenizer,
)

# Train
trainer.train()

# Save aligned model
trainer.save_model("./aligned_model")