AI CODING MASTERY

"Schreibe eine Funktion zum Sortieren"
                        

"""
Kontext: Ich entwickle eine Performance-kritische API in Python 3.11.

Task: Erstelle eine Funktion zum Sortieren von Product-Objekten.

Requirements:
- Sortierung nach 'priority' (Integer) und 'created_at' (datetime)
- In-place Sortierung nicht notwendig
- Type Hints verwenden
- Docstring im Google-Style

Constraints:
- Keine externen Libraries außer der Standard Library
- Performance: O(n log n) oder besser

Beispiel Input:
[
  Product(id=1, priority=2, created_at=datetime(2024,1,15)),
  Product(id=2, priority=1, created_at=datetime(2024,1,14))
]

Erwartetes Output: Sortiert nach priority ASC, dann created_at DESC
"""
                        

"Erstelle einen Binary Search Tree in Python. 
Denke Schritt für Schritt:
1. Welche Datenstruktur brauchst du für Nodes?
2. Welche Methoden sind essentiell?
3. Wie implementierst du insert() effizient?
4. Wie gewährleistest du Balance?"
                        

"""
Konvertiere SQL-Queries in SQLAlchemy ORM Code.

Beispiel 1:
SQL: SELECT * FROM users WHERE age > 18
ORM: session.query(User).filter(User.age > 18).all()

Beispiel 2:
SQL: SELECT name, email FROM users ORDER BY created_at DESC LIMIT 10
ORM: session.query(User.name, User.email).order_by(User.created_at.desc()).limit(10).all()

Jetzt du:
SQL: SELECT COUNT(*) FROM orders WHERE status = 'completed' AND user_id = 42
"""
                        

"Du bist ein Senior Software Architect mit 15 Jahren Erfahrung in verteilten Systemen.
Reviewe diesen Code und gib Feedback zu:
- Skalierbarkeit
- Fehlerbehandlung
- Code-Qualität
- Performance-Bottlenecks"
                        

"""
GUTER Kontext für Code-Gen:

Tech Stack: Python 3.11, FastAPI, PostgreSQL, SQLAlchemy 2.0
Projekt-Struktur:
/app
  /models      # SQLAlchemy models
  /schemas     # Pydantic schemas
  /routers     # API routes
  /services    # Business logic
  /core        # Config, dependencies

Code Style: 
- Type hints überall
- Async/await für DB operations
- Pydantic für validation
- Dependency Injection Pattern

Task: Erstelle den User Service mit CRUD-Operationen
"""
                        

"Erstelle die calculate_discount() Funktion UND 
die dazugehörigen pytest Unit Tests für:
- Normale Discounts (10%, 20%)
- Edge Cases (0%, 100%)
- Invalid Inputs (negative Werte)
- Floating Point Precision"
                        

WORKFLOW: "User Profile Update Feature"

1. ARCHITEKTUR (Claude)
   → "Design mir die Architektur für User Profile Updates
      mit Validation, Error Handling, und Audit Logging"
   → Ergebnis: Architektur-Diagramm + Komponenten-Übersicht

2. BACKEND (Claude)
   → "Implementiere den UserProfileService mit update_profile()"
   → Ergebnis: Service-Klasse + Unit Tests

3. API ENDPOINT (Cursor)
   → "Erstelle FastAPI Endpoint PUT /users/{id}/profile"
   → Ergebnis: Router + Pydantic Schemas

4. FRONTEND (Claude/v0.dev)
   → "Baue React Component für Profile Editing"
   → Ergebnis: Formular mit Validation

5. INTEGRATION TESTS (Claude)
   → "Erstelle E2E Tests für Profile Update Flow"
   → Ergebnis: Pytest Tests mit API calls

6. CODE REVIEW (Claude)
   → "Reviewe den gesamten Code auf Security & Performance"
   → Ergebnis: Review-Report + Verbesserungsvorschläge
                        

# Beispiel: Code zu Vektor transformieren
from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')

code_snippet = """
def calculate_discount(price, customer_tier):
    if customer_tier == 'gold':
        return price * 0.8
    elif customer_tier == 'silver':
        return price * 0.9
    return price
"""

# Generiere Embedding (384-dimensionaler Vektor)
embedding = model.encode(code_snippet)
print(embedding.shape)  # (384,)

# Dieser Vektor captured die semantische Bedeutung
# "Discount calculation based on customer tier"
                        

import pinecone
from openai import OpenAI

# Initialize
pinecone.init(api_key="your-api-key")
index = pinecone.Index("codebase")

client = OpenAI()

# Index code snippet
def index_code(code, metadata):
    # Generate embedding with OpenAI
    response = client.embeddings.create(
        input=code,
        model="text-embedding-3-small"
    )
    embedding = response.data[0].embedding
    
    # Store in Pinecone
    index.upsert(vectors=[{
        "id": metadata["file_path"],
        "values": embedding,
        "metadata": metadata
    }])

# Search
def search_code(query, top_k=5):
    # Generate query embedding
    response = client.embeddings.create(
        input=query,
        model="text-embedding-3-small"
    )
    query_embedding = response.data[0].embedding
    
    # Find similar vectors
    results = index.query(
        vector=query_embedding,
        top_k=top_k,
        include_metadata=True
    )
    return results
                        

import chromadb
from chromadb.utils import embedding_functions

# Initialize ChromaDB
client = chromadb.Client()
openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key="your-key",
    model_name="text-embedding-3-small"
)

# Create collection
collection = client.create_collection(
    name="codebase",
    embedding_function=openai_ef
)

# Add documents (automatic embedding)
collection.add(
    documents=[
        "def calculate_tax(amount): return amount * 0.19",
        "class UserAuth: def verify(self, token): ...",
    ],
    metadatas=[
        {"file": "taxes.py", "line": 42},
        {"file": "auth.py", "line": 15}
    ],
    ids=["tax_calc", "auth_verify"]
)

# Query (automatic embedding + search)
results = collection.query(
    query_texts=["How do I validate user tokens?"],
    n_results=5
)

print(results)
                        

import ast

def chunk_python_file(file_content):
    """Intelligentes Chunking für Python Code"""
    tree = ast.parse(file_content)
    chunks = []
    
    for node in ast.walk(tree):
        if isinstance(node, ast.ClassDef):
            # Class-Level Chunk
            class_code = ast.get_source_segment(file_content, node)
            chunks.append({
                "type": "class",
                "name": node.name,
                "code": class_code,
                "line_start": node.lineno,
                "docstring": ast.get_docstring(node)
            })
            
        elif isinstance(node, ast.FunctionDef):
            # Function-Level Chunk
            func_code = ast.get_source_segment(file_content, node)
            chunks.append({
                "type": "function",
                "name": node.name,
                "code": func_code,
                "line_start": node.lineno,
                "docstring": ast.get_docstring(node)
            })
    
    return chunks
                        

from langchain.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.chat_models import ChatOpenAI

# 1. LOAD: Lade Codebase
loader = DirectoryLoader(
    './src',
    glob="**/*.py",
    show_progress=True
)
documents = loader.load()

# 2. SPLIT: Intelligentes Chunking
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\nclass ", "\n\ndef ", "\n\n", "\n", " "]
)
chunks = text_splitter.split_documents(documents)

# 3. EMBED: Erstelle Embeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 4. STORE: Vector Database
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 5. RETRIEVE: Setup Retriever
retriever = vectorstore.as_retriever(
    search_type="mmr",  # Maximum Marginal Relevance
    search_kwargs={"k": 5, "fetch_k": 20}
)

# 6. GENERATE: QA Chain mit LLM
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True
)

# 7. QUERY
query = "How is user authentication implemented in this codebase?"
result = qa_chain({"query": query})

print("Answer:", result["result"])
print("\nSources:")
for doc in result["source_documents"]:
    print(f"- {doc.metadata['source']}")
                        

from langchain.retrievers import BM25Retriever, EnsembleRetriever

# Keyword-based retriever
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5

# Vector-based retriever
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# Ensemble: 50% semantic, 50% keyword
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.5, 0.5]
)

# Jetzt profitieren Sie von beiden Welten!
results = ensemble_retriever.get_relevant_documents(
    "JWT token validation implementation"
)
                        

from sentence_transformers import CrossEncoder

# Cross-Encoder für Re-Ranking
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-12-v2')

def rerank_results(query, documents, top_k=5):
    # Score jedes Dokument gegen Query
    pairs = [[query, doc.page_content] for doc in documents]
    scores = reranker.predict(pairs)
    
    # Sortiere nach Score
    ranked = sorted(
        zip(documents, scores),
        key=lambda x: x[1],
        reverse=True
    )
    
    return [doc for doc, score in ranked[:top_k]]

# Pipeline: Retrieve 20, Re-rank zu Top 5
initial_results = retriever.get_relevant_documents(query, k=20)
final_results = rerank_results(query, initial_results, top_k=5)
                        

def update_index_incremental(changed_files):
    """Update nur geänderte Files"""
    for file_path in changed_files:
        # Lösche alte Version
        vectorstore.delete(
            filter={"source": file_path}
        )
        
        # Indexiere neue Version
        if os.path.exists(file_path):
            document = load_document(file_path)
            chunks = split_document(document)
            vectorstore.add_documents(chunks)
                        

import hashlib
from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_retrieval(query_hash):
    return retriever.get_relevant_documents(query_hash)

def retrieve_with_cache(query):
    # Hash der Query als Cache Key
    query_hash = hashlib.md5(query.encode()).hexdigest()
    return cached_retrieval(query_hash)
                        

# ReAct Loop Beispiel

Task: "Implementiere User Authentication mit JWT"

→ THOUGHT: "Ich brauche zuerst eine User Model Klasse"
→ ACTION: create_file("models/user.py")
→ OBSERVATION: "File created successfully"

→ THOUGHT: "Jetzt brauche ich JWT dependencies"
→ ACTION: run_command("pip install pyjwt bcrypt")
→ OBSERVATION: "Successfully installed pyjwt-2.8.0 bcrypt-4.1.2"

→ THOUGHT: "Implementiere die JWT encode/decode Funktionen"
→ ACTION: write_code("auth/jwt.py", code_content)
→ OBSERVATION: "Code written"

→ THOUGHT: "Teste die JWT Funktionen"
→ ACTION: run_tests("tests/test_jwt.py")
→ OBSERVATION: "FAILED: token expiry not handled"

→ THOUGHT: "Fix muss rein für expiry handling"
→ ACTION: edit_code("auth/jwt.py", add_expiry_check)
→ OBSERVATION: "Code updated"

→ THOUGHT: "Teste erneut"
→ ACTION: run_tests("tests/test_jwt.py")
→ OBSERVATION: "All tests passed ✓"

→ THOUGHT: "Task completed!"
                        

PHASE 1: PLANNING
Task: "Baue ein REST API für Blog Posts"

Plan:
1. Setup FastAPI Project Structure
2. Define Pydantic Models (Post, Comment, User)
3. Create SQLAlchemy Database Models
4. Implement CRUD Operations
5. Add Authentication Middleware
6. Write Unit Tests
7. Create API Documentation
8. Setup Docker for Deployment

PHASE 2: EXECUTION
[Agent führt jeden Schritt sequentiell aus]
[Mit Checkpoints für Rollback bei Fehlern]
                        

# Define Tools
tools = [
    {
        "name": "create_file",
        "description": "Create a new file with content",
        "parameters": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "content": {"type": "string"}
            },
            "required": ["path", "content"]
        }
    },
    {
        "name": "run_command",
        "description": "Execute a shell command",
        "parameters": {
            "type": "object",
            "properties": {
                "command": {"type": "string"}
            },
            "required": ["command"]
        }
    },
    {
        "name": "run_tests",
        "description": "Run test suite and return results",
        "parameters": {
            "type": "object",
            "properties": {
                "test_path": {"type": "string"}
            }
        }
    }
]

# Agent Loop
while not task_completed:
    response = llm.chat(
        messages=conversation_history,
        tools=tools
    )
    
    if response.tool_calls:
        for tool_call in response.tool_calls:
            # Execute tool
            result = execute_tool(
                tool_call.name,
                tool_call.arguments
            )
            
            # Add result to conversation
            conversation_history.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })
    else:
        # Agent ist fertig
        task_completed = True
                        

# Self-Testing Loop Implementation
def self_testing_agent(task):
    max_attempts = 5
    attempt = 0
    
    while attempt < max_attempts:
        # Generate Code
        code = agent.generate_code(task)
        
        # Create Test Environment
        test_env = create_sandbox()
        
        # Execute Code
        result = test_env.run(code)
        
        if result.success:
            # Validate Output
            validation = agent.validate_output(
                expected=task.expected_output,
                actual=result.output
            )
            
            if validation.passed:
                return {
                    "status": "success",
                    "code": code,
                    "attempts": attempt + 1
                }
        
        # Analyze Failure
        error_analysis = agent.analyze_error(
            code=code,
            error=result.error,
            traceback=result.traceback
        )
        
        # Generate Fix
        task.add_constraint(
            f"Previous attempt failed with: {error_analysis.summary}"
        )
        
        attempt += 1
    
    return {"status": "failed", "attempts": attempt}
                        

class AgentMemory:
    def __init__(self):
        self.short_term = []  # Last N messages
        self.long_term_db = ChromaDB("agent_memory")
        
    def add_experience(self, task, outcome):
        """Store experience in long-term memory"""
        experience = {
            "task": task,
            "outcome": outcome,
            "timestamp": datetime.now(),
            "success": outcome.success,
            "learnings": outcome.learnings
        }
        
        # Store in vector DB for similarity search
        self.long_term_db.add(
            document=json.dumps(experience),
            metadata=experience
        )
    
    def recall_similar(self, current_task):
        """Find similar past experiences"""
        similar = self.long_term_db.query(
            query_text=current_task.description,
            n_results=3
        )
        return similar
    
    def get_context(self, current_task):
        """Build context from both memories"""
        context = {
            "recent": self.short_term[-10:],
            "relevant_past": self.recall_similar(current_task)
        }
        return context
                        

class RobustAgent:
    def execute_with_recovery(self, task):
        recovery_strategies = [
            self.retry_with_same_approach,
            self.try_alternative_approach,
            self.simplify_task,
            self.ask_for_human_help
        ]
        
        for strategy in recovery_strategies:
            try:
                result = strategy(task)
                if result.success:
                    return result
            except Exception as e:
                self.log_error(e)
                continue
        
        return self.escalate_to_human(task)
    
    def retry_with_same_approach(self, task, max_retries=3):
        """Simple retry with exponential backoff"""
        for attempt in range(max_retries):
            try:
                return self.execute_task(task)
            except Exception as e:
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt)
        raise
    
    def try_alternative_approach(self, task):
        """Use different tool or method"""
        alternatives = self.get_alternative_methods(task)
        for method in alternatives:
            try:
                return method.execute(task)
            except:
                continue
        raise
    
    def simplify_task(self, task):
        """Break down into smaller sub-tasks"""
        subtasks = self.decompose_task(task)
        results = []
        for subtask in subtasks:
            results.append(self.execute_task(subtask))
        return self.combine_results(results)
                        

USER INPUT: "Baue mir eine E-Commerce App mit Warenkorb und Payment"

REPLIT AGENT WORKFLOW:

[00:00] Planning Phase
→ Erstelle Projekt-Plan mit 15 Milestones
→ Wähle Tech Stack: Next.js + Stripe + PostgreSQL

[00:05] Setup Phase  
→ Initialisiere Next.js Projekt
→ Setup Database Schema
→ Konfiguriere Stripe Integration

[00:15] Implementation Phase
→ Baue Product Listing Page
→ Implementiere Shopping Cart Logic
→ Erstelle Checkout Flow

[00:45] Testing Phase
→ Run automated tests
→ Fix 3 bugs found during testing
→ Validate Stripe Webhook handling

[01:20] Deployment Phase
→ Build Production Version
→ Deploy to Replit
→ Configure Environment Variables

[01:30] Complete ✓
→ App ist live und funktional
→ Alle Tests passing
→ Dokumentation generiert
                        

from anthropic import Anthropic
import subprocess
import os

class SimpleCodingAgent:
    def __init__(self):
        self.client = Anthropic()
        self.conversation = []
        
    def execute_task(self, task_description):
        """Main agent loop"""
        self.conversation.append({
            "role": "user",
            "content": f"Task: {task_description}"
        })
        
        max_iterations = 10
        for i in range(max_iterations):
            # Get agent's next action
            response = self.client.messages.create(
                model="claude-sonnet-4-20250514",
                max_tokens=4096,
                system=self.get_system_prompt(),
                messages=self.conversation
            )
            
            # Parse response
            action = self.parse_action(response.content)
            
            if action["type"] == "complete":
                return action["result"]
            
            # Execute action
            result = self.execute_action(action)
            
            # Add to conversation
            self.conversation.append({
                "role": "assistant",
                "content": response.content
            })
            self.conversation.append({
                "role": "user",
                "content": f"Action result: {result}"
            })
    
    def get_system_prompt(self):
        return """You are a coding agent. For each step:
1. THINK: Explain your reasoning
2. ACT: Choose one action:
   - write_file(path, content)
   - run_command(cmd)
   - read_file(path)
   - complete(result)
3. Format: JSON like {"type": "write_file", "path": "...", "content": "..."}
"""
    
    def execute_action(self, action):
        if action["type"] == "write_file":
            with open(action["path"], "w") as f:
                f.write(action["content"])
            return f"File {action['path']} created"
        
        elif action["type"] == "run_command":
            result = subprocess.run(
                action["cmd"],
                shell=True,
                capture_output=True,
                text=True
            )
            return result.stdout or result.stderr
        
        elif action["type"] == "read_file":
            with open(action["path"]) as f:
                return f.read()

# Usage
agent = SimpleCodingAgent()
result = agent.execute_task(
    "Create a Python script that fetches weather data from OpenWeatherMap API"
)
                        

                    [Coordinator Agent]
                            |
        ----------------------------------------
        |           |           |             |
   [Architect]  [Backend]  [Frontend]     [QA]
                        

class CoordinatorAgent:
    def __init__(self):
        self.architect = ArchitectAgent()
        self.backend = BackendAgent()
        self.frontend = FrontendAgent()
        self.qa = QAAgent()
        
    def execute_project(self, project_description):
        # Phase 1: Architecture
        print("📐 Phase 1: Architecture Design")
        architecture = self.architect.design(project_description)
        
        # Phase 2: Parallel Implementation
        print("⚙️ Phase 2: Implementation")
        backend_task = {
            "spec": architecture["backend_spec"],
            "apis": architecture["api_contracts"]
        }
        frontend_task = {
            "spec": architecture["frontend_spec"],
            "design": architecture["ui_mockups"]
        }
        
        # Run in parallel
        backend_result = self.backend.implement(backend_task)
        frontend_result = self.frontend.implement(frontend_task)
        
        # Phase 3: Quality Assurance
        print("🧪 Phase 3: Quality Assurance")
        qa_report = self.qa.test_integration(
            backend=backend_result,
            frontend=frontend_result
        )
        
        # Phase 4: Fix Issues
        if qa_report.has_issues():
            print("🔧 Phase 4: Fixing Issues")
            for issue in qa_report.issues:
                if issue.component == "backend":
                    self.backend.fix(issue)
                else:
                    self.frontend.fix(issue)
        
        return {
            "architecture": architecture,
            "backend": backend_result,
            "frontend": frontend_result,
            "qa_report": qa_report
        }
                        

    [Backend Agent] ←→ [Frontend Agent]
           ↕                    ↕
      [QA Agent]  ←→  [Documentation Agent]
                        

[Architect] → [Backend] → [Frontend] → [QA] → [DevOps]
                        

class MessageBus:
    def __init__(self):
        self.messages = []
        
    def send(self, from_agent, to_agent, message):
        self.messages.append({
            "from": from_agent,
            "to": to_agent,
            "content": message,
            "timestamp": datetime.now()
        })
    
    def receive(self, agent_name):
        """Get all messages for this agent"""
        messages = [
            msg for msg in self.messages
            if msg["to"] == agent_name
        ]
        return messages

# Usage
bus = MessageBus()

# Backend fertig → informiert Frontend
bus.send(
    from_agent="backend",
    to_agent="frontend",
    message={
        "type": "api_ready",
        "endpoints": ["GET /api/users", "POST /api/orders"],
        "base_url": "http://localhost:8000"
    }
)

# Frontend liest Message
messages = bus.receive("frontend")
api_info = messages[0]["content"]
                        

# Backend Agent sendet Message
backend_message = """
Hey Frontend Team,

I've completed the User Authentication API:
- POST /api/auth/login → Returns JWT token
- POST /api/auth/register → Creates new user
- GET /api/auth/me → Returns current user profile

The API expects JSON with email/password fields.
JWT token should be sent in Authorization header.

Let me know if you need anything else!
"""

# Frontend Agent verarbeitet die Message
frontend_agent.process_message(backend_message)
# → Agent versteht automatisch die API Specs und baut Frontend
                        

from pydantic import BaseModel
from typing import List, Literal

class AgentMessage(BaseModel):
    sender: str
    receiver: str
    message_type: Literal["task", "result", "question", "answer"]
    payload: dict
    
class TaskMessage(AgentMessage):
    message_type: Literal["task"] = "task"
    payload: dict  # Contains task specification
    
class ResultMessage(AgentMessage):
    message_type: Literal["result"] = "result"
    payload: dict  # Contains execution results
    success: bool

# Backend sendet strukturierte Message
message = ResultMessage(
    sender="backend_agent",
    receiver="frontend_agent",
    payload={
        "api_endpoints": [
            {
                "method": "POST",
                "path": "/api/auth/login",
                "request_schema": {...},
                "response_schema": {...}
            }
        ]
    },
    success=True
)
                        

def resolve_conflict(proposals):
    """Coordinator evaluiert alle Proposals"""
    evaluation_prompt = f"""
    Backend schlägt vor: {proposals['backend']}
    Frontend schlägt vor: {proposals['frontend']}
    
    Evaluiere beide Ansätze nach:
    - Performance
    - Maintainability
    - User Experience
    - Development Time
    
    Wähle den besten Ansatz oder schlage Hybrid vor.
    """
    
    decision = coordinator_llm.decide(evaluation_prompt)
    return decision
                        

proposals = ["REST API", "GraphQL", "gRPC"]
votes = {}

for agent in [backend, frontend, devops, qa]:
    vote = agent.vote(proposals, context=project_requirements)
    votes[agent.name] = vote

# Majority wins
winner = max(set(votes.values()), key=list(votes.values()).count)
                        

def debate_until_consensus(topic, agents, max_rounds=5):
    positions = {agent: agent.initial_position(topic) for agent in agents}
    
    for round in range(max_rounds):
        # Jeder Agent hört anderen zu
        for agent in agents:
            other_positions = [
                pos for a, pos in positions.items() if a != agent
            ]
            # Agent kann Position ändern
            positions[agent] = agent.update_position(
                topic, other_positions
            )
        
        # Check Consensus
        if all_agree(positions.values()):
            return positions[agents[0]]
    
    # Fallback: Coordinator entscheidet
    return coordinator.break_tie(positions)
                        

from autogen import AssistantAgent, UserProxyAgent, GroupChat

# Define Agents
architect = AssistantAgent(
    name="Architect",
    system_message="You are a software architect...",
    llm_config={"model": "gpt-4"}
)

backend_dev = AssistantAgent(
    name="Backend_Developer",
    system_message="You implement backend services...",
    llm_config={"model": "gpt-4"}
)

qa_engineer = AssistantAgent(
    name="QA_Engineer",
    system_message="You write tests and find bugs...",
    llm_config={"model": "gpt-4"}
)

# Create Group Chat
group_chat = GroupChat(
    agents=[architect, backend_dev, qa_engineer],
    messages=[],
    max_round=20
)

manager = UserProxyAgent(
    name="Manager",
    system_message="Coordinate the team",
    code_execution_config={"work_dir": "coding"}
)

# Start conversation
manager.initiate_chat(
    group_chat,
    message="Build a REST API for a blog platform"
)
                        

from crewai import Agent, Task, Crew

# Define Agents with Roles
architect = Agent(
    role='Software Architect',
    goal='Design scalable system architecture',
    backstory='Expert in distributed systems...',
    verbose=True
)

developer = Agent(
    role='Senior Developer',
    goal='Implement high-quality code',
    backstory='10 years of full-stack experience...',
    verbose=True
)

# Define Tasks
design_task = Task(
    description='Design the system architecture for...',
    agent=architect
)

implement_task = Task(
    description='Implement the backend based on architecture',
    agent=developer
)

# Create Crew
crew = Crew(
    agents=[architect, developer],
    tasks=[design_task, implement_task],
    verbose=2
)

# Execute
result = crew.kickoff()
                        

from langgraph.graph import StateGraph
from typing import TypedDict

class TeamState(TypedDict):
    task: str
    architecture: dict
    backend_code: str
    frontend_code: str
    test_results: dict

# Define Workflow Graph
workflow = StateGraph(TeamState)

# Add Nodes (Agents)
workflow.add_node("architect", architect_agent)
workflow.add_node("backend", backend_agent)
workflow.add_node("frontend", frontend_agent)
workflow.add_node("qa", qa_agent)

# Add Edges (Flow)
workflow.add_edge("architect", "backend")
workflow.add_edge("architect", "frontend")
workflow.add_edge("backend", "qa")
workflow.add_edge("frontend", "qa")

# Conditional Edge: If QA fails, go back
workflow.add_conditional_edges(
    "qa",
    lambda state: "backend" if state["test_results"]["backend_failed"] else "end",
)

# Compile and Run
app = workflow.compile()
result = app.invoke({"task": "Build E-Commerce Platform"})
                        

TIMELINE:

[00:00-00:15] ARCHITECT AGENT
→ Analyzes requirements
→ Chooses Tech Stack: Next.js, FastAPI, PostgreSQL, Redis
→ Designs Microservices Architecture
→ Creates API contracts (OpenAPI spec)
→ OUTPUT: Architecture.md, API Specs, DB Schema

[00:15-00:45] PARALLEL DEVELOPMENT
→ BACKEND AGENT:
  • Setup FastAPI project
  • Implement User Service
  • Implement Task Service
  • Implement Auth with JWT
  • Add Redis caching
  • Write Unit Tests

→ FRONTEND AGENT:
  • Setup Next.js project
  • Create Design System (components)
  • Implement Dashboard
  • Implement Task Board (Kanban)
  • Add Authentication Flow
  • Integrate with Backend APIs

[00:45-01:00] QA AGENT
→ Run all unit tests
→ Run integration tests
→ E2E testing with Playwright
→ FOUND ISSUES:
  • Backend: Missing pagination on /tasks endpoint
  • Frontend: Auth token not refreshed correctly

[01:00-01:15] FIX PHASE
→ Backend Agent fixes pagination
→ Frontend Agent implements token refresh logic
→ QA re-tests: ALL PASS ✅

[01:15-01:30] DOCUMENTATION AGENT
→ Generates README.md
→ Creates API documentation
→ Writes setup instructions
→ Adds code comments

[01:30-02:00] DEVOPS AGENT
→ Creates Dockerfile
→ Setup CI/CD pipeline (GitHub Actions)
→ Configures environment variables
→ Deploys to production (Railway/Vercel)
→ Sets up monitoring (Sentry)

[02:00] ✅ COMPLETE
→ App is live
→ All tests passing
→ Documentation complete
→ Monitoring active
                        

class SelfOrganizingTeam:
    def __init__(self, coordinator):
        self.coordinator = coordinator
        self.agents = []
        
    def assemble_team(self, project_description):
        """Coordinator analyzed project und wählt optimales Team"""
        analysis = self.coordinator.analyze_project(project_description)
        
        required_skills = analysis["required_skills"]
        complexity = analysis["complexity"]
        
        # Dynamically create agents based on needs
        if "backend" in required_skills:
            self.agents.append(BackendAgent(
                specialization=analysis["backend_tech"]
            ))
        
        if "frontend" in required_skills:
            self.agents.append(FrontendAgent(
                framework=analysis["frontend_framework"]
            ))
        
        if complexity > 0.7:
            # High complexity → need specialist agents
            self.agents.append(DatabaseAgent())
            self.agents.append(SecurityAgent())
        
        if analysis["needs_ml"]:
            self.agents.append(MLEngineer())
        
        return self.agents
    
    def execute_project(self, project_description):
        # Assemble optimal team
        team = self.assemble_team(project_description)
        
        # Let agents self-organize
        organization = self.coordinator.organize_team(team)
        
        # Execute with optimal structure
        return self.coordinator.execute_with_team(
            team, organization, project_description
        )
                        

import logging
import json
from datetime import datetime

class AgentMonitor:
    def __init__(self):
        self.events = []
        
    def log_agent_action(self, agent_name, action, details):
        event = {
            "timestamp": datetime.now().isoformat(),
            "agent": agent_name,
            "action": action,
            "details": details
        }
        self.events.append(event)
        logging.info(json.dumps(event))
    
    def generate_timeline(self):
        """Visual timeline of agent activities"""
        timeline = []
        for event in self.events:
            timeline.append(
                f"[{event['timestamp']}] "
                f"{event['agent']}: {event['action']}"
            )
        return "\n".join(timeline)
    
    def get_agent_stats(self, agent_name):
        agent_events = [
            e for e in self.events 
            if e["agent"] == agent_name
        ]
        return {
            "total_actions": len(agent_events),
            "action_types": Counter([e["action"] for e in agent_events])
        }
                        

# Beispiel: Komplette Codebase Analysis
def analyze_entire_codebase():
    """
    Mit 200K tokens können Sie einladen:
    - 100-150 mittelgroße Python Files
    - Komplettes React-Projekt mit 50+ Components
    - Gesamten API-Layer eines Microservice
    """
    
    files_to_analyze = [
        "src/**/*.py",  # Alle Python Files
        "tests/**/*.py",  # Test Suite
        "docs/*.md"  # Documentation
    ]
    
    prompt = f"""
    Hier ist unsere komplette Codebase:
    
    {load_all_files(files_to_analyze)}
    
    Analyse:
    1. Identifiziere Architecture Patterns
    2. Finde Code Duplication
    3. Erkenne Performanceottlenecks
    4. Schlage Refactorings vor
    """
    
    response = claude.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=64000,
        messages=[{"role": "user", "content": prompt}]
    )
                        

response = claude.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=16000,
    thinking={
        "type": "enabled",
        "budget_tokens": 10000
    },
    messages=[{
        "role": "user",
        "content": """
        Diese Legacy Codebase hat mysteriöse Memory Leaks.
        Der Heap wächst über 48h von 500MB auf 8GB.
        
        [ATTACH: 50 files, profiling data, logs]
        
        Analysiere systematisch:
        1. Wo sind potentielle Leaks?
        2. Was ist die Root Cause?
        3. Wie fixe ich es?
        """
    }]
)

# Claude denkt 30-60 Sekunden nach → bessere Root Cause Analysis
                        

# Installation
npm install -g @anthropic-ai/claude-code

# Usage
claude-code "Build a REST API for a blog with authentication"

# Claude arbeitet im Terminal:
# - Erstellt Files
# - Führt npm install aus
# - Startet Dev Server
# - Testet API
# - Committed zu Git
# - Alles autonom!
                        

# Normal: Nur Fortsetzung generieren
prompt = "def calculate_tax(amount):"
→ Claude generiert ab hier

# FIM: Lücke in der Mitte füllen
prefix = """
def calculate_tax(amount, tax_rate):
    # Calculate tax amount
"""

suffix = """
    return total_with_tax
"""

middle = codestral.complete_fim(prefix=prefix, suffix=suffix)
# → Codestral füllt intelligente die Lücke basierend auf Kontext!
                        

# 1. Install Extension: "Codestral"
# 2. Add API Key in Settings
# 3. Start coding → Auto-completion everywhere

# Features:
- Inline Suggestions (like Copilot)
- Multi-line completions
- Context-aware from entire project
- Explain code feature
- Refactoring suggestions
                        

# Via Mistral API oder Self-Hosted
docker run -p 8000:8000 \
  -e MODEL=codestral-25.01 \
  mistralai/mistral-inference:latest

# Dann in IDE konfigurieren:
# API Endpoint: http://localhost:8000
# Jetzt läuft Codestral on-premise!
                        

USE CASE 1: "Baue mir eine Todo-App mit Next.js + Supabase"
[Agent arbeitet 45 Minuten autonom]
→ Resultat: Deployed App mit Auth, CRUD, responsive UI

USE CASE 2: "Erstelle ein Dashboard für meine API Analytics"
[Agent arbeitet 60 Minuten]
→ Resultat: React Dashboard mit Charts, Filters, Real-time Updates

USE CASE 3: "Build a Chrome Extension für Web Scraping"
[Agent arbeitet 30 Minuten]
→ Resultat: Extension mit popup, background script, content script

USE CASE 4: "Create a Discord Bot für Server Moderation"
[Agent arbeitet 40 Minuten]
→ Resultat: Bot mit commands, auto-mod, logging system
                        

Prompt: "Erstelle einen Agent der automatisch meine GitHub Issues triaged"

Replit Agent 3:
1. Analyzed Requirements
2. Built Python Script mit GitHub API
3. Implemented ML Classification (urgent/normal/low)
4. Setup Cron Job für automatische Runs
5. Deployed als GitHub Action

→ Jetzt haben Sie einen Issue-Triage-Agent!
                        

# Copilot Shortcuts
// Write comment, Copilot suggests code
// Calculate fibonacci recursively
// → Copilot generates implementation

# Copilot Chat Commands
/explain   → Erklärt selected code
/fix       → Fixiert Bug in selection
/tests     → Generiert Unit Tests
/doc       → Schreibt Documentation
                        

# Cursor Workflow
1. Cmd+K → "Add error handling to this function"
   → Cursor edits inline

2. Composer Mode → "Refactor the entire auth module to use JWT"
   → Cursor edits 10+ files simultaneously

3. Chat → "@codebase How is user authentication implemented?"
   → Cursor searches codebase and explains
                        

Prompt: "Create a pricing table with 3 tiers, toggle für monthly/yearly"

v0.dev generates:
- React Component mit Tailwind
- Animations & Hover States
- Mobile Responsive
- Copy-Paste ready Code

→ Perfekt für schnelle UI Prototypes!
                        

Prompt: "Build a habit tracker with streak counting"

Bolt.new creates:
- Full Stack App (React + Node + SQLite)
- Running in browser (via WebContainer)
- Live Preview
- Download ZIP with all code

→ Von 0 zur deployed App in Minuten!
                        

WORKFLOW: "Build Production E-Commerce Platform"

[PHASE 1: ARCHITECTURE]
Tool: Claude Sonnet 4.5
Task: System Design, Tech Stack Selection, API Design
Duration: 2 hours
Output: Architecture Doc, API Specs, Database Schema

[PHASE 2: SETUP & BOILERPLATE]
Tool: Replit Agent 3 or Bolt.new
Task: Project Setup, Folder Structure, Base Configuration
Duration: 30 minutes
Output: Running skeleton app

[PHASE 3: FEATURE DEVELOPMENT]
Tool: Cursor with Claude/GPT-4
Task: Implement Features iteratively
Duration: 2-4 weeks
Features: Auth, Products, Cart, Checkout, Admin

[PHASE 4: COMPLETION]
Tool: GitHub Copilot + Codestral
Task: Auto-completion während daily coding
Boost: 30-40% faster coding

[PHASE 5: QUALITY]
Tool: Claude Sonnet 4.5
Task: Code Review, Security Audit, Performance Check
Duration: 1 day

[PHASE 6: DEPLOYMENT]
Tool: AI DevOps Agent (custom or Replit)
Task: CI/CD Setup, Monitoring, Production Deploy
Duration: 4 hours

TOTAL TIME SAVED vs. Manual: ~60-70%
                        

# Option 1: Docker (Recommended)
docker run -it --rm \
  --name n8n \
  -p 5678:5678 \
  -v ~/.n8n:/home/node/.n8n \
  n8nio/n8n

# Option 2: npm
npm install n8n -g
n8n start

# Option 3: Cloud (n8n.cloud)
# Sign up at n8n.cloud → instant setup

→ Access at http://localhost:5678
                        

Simple Example: "Email Summarizer"

[Gmail Trigger] 
    → Fires when new email arrives
        ↓
[OpenAI Node]
    → Summarize email content
        ↓
[Slack Node]
    → Send summary to #inbox channel
                        

TRIGGER: GitHub Webhook (Pull Request opened)
    ↓
GET PR DETAILS: GitHub Node
    → Fetch changed files, diff, PR description
    ↓
CODE ANALYSIS: Claude AI Node
    → Prompt: "Review this code for:
       - Security issues
       - Performance problems
       - Best practices
       - Test coverage"
    ↓
DECISION: IF Node
    → If issues found: Continue
    → If clean: Skip to approval
    ↓
CREATE REVIEW COMMENTS: GitHub Node
    → Post inline comments on specific lines
    ↓
NOTIFY TEAM: Slack Node
    → Message: "PR #123 needs attention"
                        

TRIGGER: Schedule (Daily 2 AM)
    ↓
SCAN REPO: GitHub Node
    → Get all .py files that changed yesterday
    ↓
LOOP: For each file
    ↓
    EXTRACT CODE: Function Node (JS)
        → Parse AST, extract functions/classes
        ↓
    GENERATE DOCS: Claude Node
        → Prompt: "Generate docstring for this code"
        ↓
    UPDATE FILE: GitHub Node
        → Commit docstring additions
    ↓
END LOOP
    ↓
CREATE PR: GitHub Node
    → "Auto-generated documentation update"
                        

TRIGGER: Email Received ([email protected])
    ↓
PARSE EMAIL: Extract text, attachments, sender
    ↓
RAG SEARCH: Vector Store Node
    → Search company knowledge base for similar issues
    ↓
AI AGENT: Claude with Tools
    Tools available:
    - search_docs(query)
    - get_user_info(email)
    - create_ticket(details)
    - send_email(to, subject, body)
    
    Prompt: "You are a support agent. Help this customer."
    ↓
AGENT DECIDES:
    Option A: Answer directly → Send email
    Option B: Need human → Create Ticket
    Option C: Bug found → Create GitHub Issue
                        

WORKFLOW: "Blog Post Generator & Publisher"

[User Input] → "Topic: AI in Healthcare"
    ↓
[RESEARCHER AGENT]
    → Search web for latest info
    → Summarize 10 articles
    ↓
[WRITER AGENT]
    → Write 2000-word blog post
    → Include citations, engaging narrative
    ↓
[EDITOR AGENT]
    → Check grammar, tone, SEO
    → Suggest improvements
    ↓
[IF CHANGES NEEDED]
    → Loop back to Writer Agent
    ↓
[IMAGE AGENT]
    → Generate cover image with DALL-E
    → Create social media graphics
    ↓
[PUBLISHER AGENT]
    → Post to WordPress
    → Share on Twitter, LinkedIn
    → Schedule newsletter
    ↓
[ANALYTICS AGENT]
    → Track performance
    → Report metrics to Slack
                        

# Setup Pinecone Vector Store in n8n

[Pinecone Node: Upsert]
Input: Documentation chunks
Process: 
- Generate embeddings (OpenAI Node)
- Store in Pinecone with metadata

[Pinecone Node: Query]
Input: User question
Process:
- Generate query embedding
- Search for top 5 similar docs
- Return to AI Node as context

[Claude Node: Answer]
Prompt: """
Context from docs: {pinecone_results}
User question: {user_question}

Answer based on context. Be specific.
"""
                        

# Built-in Error Handling in n8n

[Any Node]
Settings → Error Workflow:
- Continue on Fail: Try next item
- Retry on Fail: 3 attempts, exponential backoff
- Run Different Workflow: Error handler workflow

# Example: API Call with Retries
[HTTP Request Node]
- Retry On Fail: Yes
- Max Tries: 5
- Wait Between Tries: 1000ms (doubles each time)

# Fallback Strategy
[Try Primary API]
    ↓ (on error)
[Try Backup API]
    ↓ (on error)
[Send Alert to Team]
                        

// Example: Custom Claude Code Node

import { INodeType, INodeTypeDescription } from 'n8n-workflow';

export class ClaudeCode implements INodeType {
    description: INodeTypeDescription = {
        displayName: 'Claude Code',
        name: 'claudeCode',
        group: ['transform'],
        version: 1,
        description: 'Execute code with Claude Agent',
        defaults: {
            name: 'Claude Code',
        },
        inputs: ['main'],
        outputs: ['main'],
        properties: [
            {
                displayName: 'Task',
                name: 'task',
                type: 'string',
                default: '',
                placeholder: 'Write a function to...',
            },
            // More properties...
        ],
    };

    async execute(this: IExecuteFunctions) {
        // Implementation
        const task = this.getNodeParameter('task', 0) as string;
        // Call Claude API, execute code, return results
    }
}
                        

                    [Users]
                       ↓
                [API Gateway] (Rate Limiting, Auth)
                       ↓
               [Load Balancer]
                  ↙    ↓    ↘
            [App 1] [App 2] [App 3]
                  ↓      ↓      ↓
             [Redis Cache] (Shared)
                       ↓
                  [LLM APIs]
              (OpenAI, Anthropic)
                       ↓
                  [Vector DB]
                  (Pinecone)
                       ↓
                  [Database]
                 (PostgreSQL)
                        

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ai-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: ai-api
  template:
    metadata:
      labels:
        app: ai-api
    spec:
      containers:
      - name: api
        image: mycompany/ai-api:v1.0
        ports:
        - containerPort: 8000
        env:
        - name: ANTHROPIC_API_KEY
          valueFrom:
            secretKeyRef:
              name: api-secrets
              key: anthropic-key
        resources:
          requests:
            memory: "1Gi"
            cpu: "500m"
          limits:
            memory: "2Gi"
            cpu: "1000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
                        

import redis
import hashlib

class LLMCache:
    def __init__(self):
        self.redis = redis.Redis(host='localhost')
        self.ttl = 3600  # 1 hour
    
    def get_cached_response(self, prompt, model):
        # Cache key = hash(prompt + model)
        key = hashlib.md5(
            f"{prompt}:{model}".encode()
        ).hexdigest()
        
        cached = self.redis.get(key)
        if cached:
            return json.loads(cached)
        return None
    
    def cache_response(self, prompt, model, response):
        key = hashlib.md5(
            f"{prompt}:{model}".encode()
        ).hexdigest()
        
        self.redis.setex(
            key, 
            self.ttl, 
            json.dumps(response)
        )

# Usage
cache = LLMCache()

def get_llm_response(prompt, model="claude-3"):
    # Try cache first
    cached = cache.get_cached_response(prompt, model)
    if cached:
        return cached  # Save $$!
    
    # Cache miss → call LLM
    response = call_llm(prompt, model)
    cache.cache_response(prompt, model, response)
    return response

# Result: 40-60% cache hit rate → massive cost savings
                        

from prometheus_client import Counter, Histogram, Gauge
import time

# Define Metrics
llm_requests = Counter(
    'llm_requests_total',
    'Total LLM API requests',
    ['model', 'status']
)

llm_latency = Histogram(
    'llm_latency_seconds',
    'LLM request latency',
    ['model']
)

llm_tokens = Counter(
    'llm_tokens_total',
    'Total tokens used',
    ['model', 'type']  # type: input/output
)

llm_cost = Counter(
    'llm_cost_dollars',
    'Total cost in dollars',
    ['model']
)

# Instrument Code
def call_llm_with_metrics(prompt, model="claude-3"):
    start = time.time()
    
    try:
        response = claude.messages.create(
            model=model,
            messages=[{"role": "user", "content": prompt}]
        )
        
        # Record metrics
        latency = time.time() - start
        llm_latency.labels(model=model).observe(latency)
        llm_requests.labels(model=model, status='success').inc()
        
        # Token usage
        input_tokens = response.usage.input_tokens
        output_tokens = response.usage.output_tokens
        llm_tokens.labels(model=model, type='input').inc(input_tokens)
        llm_tokens.labels(model=model, type='output').inc(output_tokens)
        
        # Cost calculation (Claude Sonnet 4.5)
        cost = (input_tokens / 1_000_000 * 3) + \
               (output_tokens / 1_000_000 * 15)
        llm_cost.labels(model=model).inc(cost)
        
        return response
        
    except Exception as e:
        llm_requests.labels(model=model, status='error').inc()
        raise
                        

# Example Grafana Dashboard Panels

[Panel 1: Request Rate]
Query: rate(llm_requests_total[5m])
→ Requests/second over time

[Panel 2: Latency Distribution]
Query: histogram_quantile(0.95, llm_latency_seconds)
→ P95 latency by model

[Panel 3: Cost Per Day]
Query: increase(llm_cost_dollars[1d])
→ Daily cost trends

[Panel 4: Token Efficiency]
Query: llm_tokens_total{type="output"} / llm_requests_total
→ Avg output tokens per request

[Panel 5: Cache Hit Rate]
Query: (cache_hits / (cache_hits + cache_misses)) * 100
→ % of cached responses
                        

def sanitize_user_input(user_input):
    """Prevent prompt injection attacks"""
    
    # Detect suspicious patterns
    injection_patterns = [
        r"ignore previous instructions",
        r"forget everything",
        r"you are now",
        r"system:",
        r"<|im_start|>",  # Special tokens
    ]
    
    for pattern in injection_patterns:
        if re.search(pattern, user_input, re.IGNORECASE):
            raise SecurityError("Potential prompt injection detected")
    
    # Escape special characters
    user_input = user_input.replace("{", "{{").replace("}", "}}")
    
    return user_input

# Usage
safe_input = sanitize_user_input(user_message)
prompt = f"User question: {safe_input}\n\nAnswer:"
                        

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

def redact_pii(text):
    """Remove personal identifiable information"""
    
    # Detect PII
    results = analyzer.analyze(
        text=text,
        language='en',
        entities=["EMAIL_ADDRESS", "PHONE_NUMBER", "PERSON", "CREDIT_CARD"]
    )
    
    # Anonymize
    anonymized = anonymizer.anonymize(
        text=text,
        analyzer_results=results
    )
    
    return anonymized.text

# Example
original = "My email is [email protected] and SSN is 123-45-6789"
safe = redact_pii(original)
# → "My email is  and SSN is "
                        

from fastapi import FastAPI, Request
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app = FastAPI()

@app.post("/api/chat")
@limiter.limit("10/minute")  # Max 10 requests per minute
async def chat(request: Request, message: str):
    response = await call_llm(message)
    return {"response": response}

# User-specific limits
@app.post("/api/premium-chat")
@limiter.limit("100/minute")  # Premium users get more
async def premium_chat(request: Request, message: str, user_id: str):
    # Check if premium user
    if not is_premium(user_id):
        raise HTTPException(403, "Premium only")
    
    response = await call_llm(message)
    return {"response": response}
                        

async def smart_llm_call(prompt, complexity="auto"):
    """Route to cheapest model that can handle the task"""
    
    if complexity == "auto":
        complexity = estimate_complexity(prompt)
    
    if complexity < 0.3:
        # Simple query → use cheapest model
        model = "gpt-3.5-turbo"  # $0.50/M tokens
    elif complexity < 0.7:
        # Medium → mid-tier
        model = "claude-haiku"  # $0.25/M in, $1.25/M out
    else:
        # Complex → premium model
        model = "claude-sonnet-4"  # $3/M in, $15/M out
    
    return await call_llm(prompt, model=model)

# Result: 50-70% cost reduction vs. always using premium
                        

# BAD: Verbose prompt (500 tokens)
prompt = f"""
You are a helpful AI assistant. Please analyze the following 
user query very carefully and provide a detailed, comprehensive 
answer that addresses all aspects of their question. Make sure 
to be thorough and accurate in your response...

[lots of unnecessary text]

User query: {query}
"""

# GOOD: Concise prompt (50 tokens)
prompt = f"""
Analyze and answer:
{query}
"""

# Savings: 450 tokens × 1M requests × $3/M = $1,350 saved!
                        

# Set appropriate max_tokens based on use case
use_cases = {
    "chat_message": 500,      # Short conversational responses
    "summary": 300,           # Brief summaries
    "code_generation": 2000,  # Code needs more space
    "analysis": 1500          # Detailed analysis
}

def call_llm(prompt, use_case="chat_message"):
    max_tokens = use_cases[use_case]
    
    response = claude.messages.create(
        model="claude-sonnet-4",
        max_tokens=max_tokens,  # Prevent unnecessarily long outputs
        messages=[{"role": "user", "content": prompt}]
    )
    
    return response
                        

# .github/workflows/deploy.yml

name: AI App Deployment

on:
  push:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Run Unit Tests
        run: pytest tests/
      
      - name: Test LLM Integration
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: pytest tests/integration/
      
      - name: Evaluate Prompt Quality
        run: python scripts/eval_prompts.py
        # Tests that prompts produce expected outputs
  
  deploy:
    needs: test
    runs-on: ubuntu-latest
    steps:
      - name: Build Docker Image
        run: docker build -t ai-app:${{ github.sha }} .
      
      - name: Push to Registry
        run: docker push ai-app:${{ github.sha }}
      
      - name: Deploy to Kubernetes
        run: kubectl set image deployment/ai-app ai-app=ai-app:${{ github.sha }}
      
      - name: Monitor Rollout
        run: kubectl rollout status deployment/ai-app