ai-dev-guidelines

# AI/ML Development Guidelines (LangChain/LangGraph/FastAPI)

## Purpose

Establish best practices for integrating AI/ML capabilities into FastAPI applications, with focus on LangChain, LangGraph, and ML model deployment.

## When to Use This Skill

Automatically activates when working on:

- LangChain chains, agents, tools
- LangGraph workflows and state machines
- RAG (Retrieval Augmented Generation)
- Prompt engineering and templates
- Vector stores and embeddings
- ML model integration (sentiment analysis, aspect-based analysis, etc.)
- Streaming LLM responses
- AI service layers
- Model deployment and optimization

---

## Quick Start

### New AI Feature Checklist

- [ ] **Model Selection**: Choose appropriate model/API (OpenAI, Anthropic, local)
- [ ] **Prompt Design**: Create prompt templates
- [ ] **Chain/Graph**: Build LangChain chain or LangGraph workflow
- [ ] **API Endpoint**: FastAPI route with streaming support
- [ ] **Error Handling**: Retry logic, fallbacks, timeouts
- [ ] **Validation**: Input/output validation with Pydantic
- [ ] **Monitoring**: Log tokens, latency, errors
- [ ] **Testing**: Unit tests with mocked LLM responses
- [ ] **Documentation**: Document prompts and expected behavior

---

## Architecture for AI Features

### Layered AI Architecture

```
HTTP Request
    ↓
FastAPI Route (streaming setup)
    ↓
AI Service (chain/graph orchestration)
    ↓
LangChain/LangGraph (LLM calls)
    ↓
Model/API (OpenAI, Anthropic, HuggingFace, local)
```

**Current Project ML Structure:**

```
backend/app/
├── routes/
│   └── ML_Routes.py        # ML API endpoints
├── machine_learning/
│   └── mlendpoint.py       # Sentiment analysis functions
└── services/               # TO BE CREATED for LangChain/LangGraph
    └── ai_services/
        ├── sentiment.py    # Refactored sentiment analysis
        ├── chains.py       # LangChain chains
        ├── graphs.py       # LangGraph workflows
        └── prompts.py      # Prompt templates
```

---

## Package Management

**This project uses `uv` for Python package management.**

```bash
# Add LangChain dependencies
uv add langchain langchain-openai langchain-anthropic langgraph

# Add vector store dependencies (when needed)
uv add faiss-cpu chromadb

# Add other AI dependencies
uv add tiktoken sentence-transformers

# Run Python scripts with uv
uv run python script.py
```

**❌ NEVER use `pip install`** - Always use `uv add` instead.

---

## Core Principles for AI Development

### 1. Separate Prompts from Code

```python
# ❌ NEVER: Hardcoded prompts in code
def analyze_text(text: str):
    response = llm.invoke(f"Analyze sentiment of: {text}")
    return response

# ✅ ALWAYS: Templated prompts
from langchain.prompts import ChatPromptTemplate

SENTIMENT_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "You are a sentiment analysis expert."),
    ("user", "Analyze the sentiment of the following text:\n\n{text}\n\nProvide: sentiment (positive/neutral/negative) and confidence (0-1).")
])

def analyze_text(text: str):
    chain = SENTIMENT_PROMPT | llm | output_parser
    return chain.invoke({"text": text})
```

### 2. Use Pydantic for LLM Output Validation

```python
from pydantic import BaseModel, Field
from langchain.output_parsers import PydanticOutputParser

class SentimentResult(BaseModel):
    sentiment: str = Field(description="positive, neutral, or negative")
    confidence: float = Field(ge=0.0, le=1.0)
    reasoning: str = Field(description="Brief explanation")

# Create parser
parser = PydanticOutputParser(pydantic_object=SentimentResult)

# Add format instructions to prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a sentiment expert."),
    ("user", "{text}\n\n{format_instructions}")
])

# Use in chain
chain = prompt | llm | parser

result: SentimentResult = chain.invoke({
    "text": "I love this product!",
    "format_instructions": parser.get_format_instructions()
})
```

### 3. Handle Streaming for Better UX

```python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

# Setup streaming LLM
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4",
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)

@app.post("/ai/generate-stream")
async def generate_stream(request: GenerateRequest):
    """Stream LLM responses for real-time feedback"""

    async def event_generator():
        async for chunk in chain.astream({"input": request.text}):
            if chunk:
                yield f"data: {chunk}\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream"
    )
```

### 4. Implement Retry Logic and Fallbacks

```python
from tenacity import retry, stop_after_attempt, wait_exponential
from langchain.chat_models import ChatOpenAI, ChatAnthropic

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def call_llm_with_retry(prompt: str):
    try:
        # Try primary model
        response = await primary_llm.ainvoke(prompt)
        return response
    except Exception as e:
        # Log and fallback
        logger.error(f"Primary LLM failed: {e}")
        # Try fallback model
        response = await fallback_llm.ainvoke(prompt)
        return response
```

### 5. Use Environment Variables for API Keys

```python
# ❌ NEVER: Hardcoded API keys
llm = ChatOpenAI(api_key="sk-...")

# ✅ ALWAYS: From environment
import os
from pydantic_settings import BaseSettings

class AISettings(BaseSettings):
    openai_api_key: str
    anthropic_api_key: str
    model_name: str = "gpt-4"
    max_tokens: int = 1000
    temperature: float = 0.7

    class Config:
        env_file = ".env"

settings = AISettings()

llm = ChatOpenAI(
    api_key=settings.openai_api_key,
    model=settings.model_name,
    max_tokens=settings.max_tokens
)
```

### 6. Track Token Usage and Costs

```python
from langchain.callbacks import get_openai_callback

@app.post("/ai/analyze")
async def analyze_with_tracking(text: str):
    with get_openai_callback() as cb:
        result = chain.invoke({"text": text})

        # Log metrics
        logger.info(f"""
        Tokens used: {cb.total_tokens}
        Prompt tokens: {cb.prompt_tokens}
        Completion tokens: {cb.completion_tokens}
        Total cost: ${cb.total_cost}
        """)

    return {
        "result": result,
        "metadata": {
            "tokens": cb.total_tokens,
            "cost": cb.total_cost
        }
    }
```

### 7. Implement Proper Error Boundaries

```python
from fastapi import HTTPException
from langchain.schema import LLMResult
from openai.error import RateLimitError, APIError

@app.post("/ai/process")
async def process_with_ai(request: AIRequest):
    try:
        result = await ai_service.process(request.text)
        return result

    except RateLimitError:
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded. Please try again later."
        )

    except APIError as e:
        logger.error(f"LLM API error: {e}")
        raise HTTPException(
            status_code=503,
            detail="AI service temporarily unavailable"
        )

    except Exception as e:
        logger.error(f"Unexpected error in AI processing: {e}")
        raise HTTPException(
            status_code=500,
            detail="Error processing request"
        )
```

---

## LangChain Patterns

### Basic Chain Pattern

```python
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import StrOutputParser

# Components
llm = ChatOpenAI(model="gpt-4")
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{input}")
])
output_parser = StrOutputParser()

# Chain
chain = prompt | llm | output_parser

# Invoke
result = chain.invoke({"input": "Hello!"})
```

### Chain with Memory

```python
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

memory = ConversationBufferMemory()

conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# First interaction
response1 = conversation.predict(input="Hi, I'm Aaron")

# Memory persists
response2 = conversation.predict(input="What's my name?")
# Will respond: "Your name is Aaron"
```

### RAG Chain Pattern

```python
from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA

# 1. Prepare documents
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
texts = text_splitter.split_documents(documents)

# 2. Create vector store
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(texts, embeddings)

# 3. Create retrieval chain
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3})
)

# 4. Query
result = qa_chain.invoke({"query": "What are the main findings?"})
```

### Tool-Using Agent

```python
from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain.tools import Tool

# Define tools
def search_projects(query: str) -> str:
    """Search projects in database"""
    # Your search logic
    return f"Found 3 projects matching {query}"

def get_sentiment(text: str) -> str:
    """Analyze sentiment of text"""
    # Your sentiment analysis
    return "positive"

tools = [
    Tool(
        name="search_projects",
        func=search_projects,
        description="Search for urban planning projects by name or city"
    ),
    Tool(
        name="analyze_sentiment",
        func=get_sentiment,
        description="Analyze sentiment of text (positive/neutral/negative)"
    )
]

# Create agent
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

# Execute
result = agent_executor.invoke({
    "input": "Find projects in Mexico City and analyze sentiment"
})
```

---

## LangGraph Patterns

### Simple State Machine

```python
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from operator import add

# Define state
class AgentState(TypedDict):
    input: str
    analysis: str
    sentiment: str
    final_output: str

# Define nodes
def analyze_node(state: AgentState):
    # Perform analysis
    analysis = llm.invoke(f"Analyze: {state['input']}")
    return {"analysis": analysis}

def sentiment_node(state: AgentState):
    # Extract sentiment
    sentiment = extract_sentiment(state['analysis'])
    return {"sentiment": sentiment}

def format_node(state: AgentState):
    # Format output
    output = f"Analysis: {state['analysis']}\nSentiment: {state['sentiment']}"
    return {"final_output": output}

# Build graph
workflow = StateGraph(AgentState)

workflow.add_node("analyze", analyze_node)
workflow.add_node("sentiment", sentiment_node)
workflow.add_node("format", format_node)

workflow.set_entry_point("analyze")
workflow.add_edge("analyze", "sentiment")
workflow.add_edge("sentiment", "format")
workflow.add_edge("format", END)

app = workflow.compile()

# Run
result = app.invoke({"input": "This project is amazing!"})
```

### Conditional Routing

```python
def route_by_sentiment(state: AgentState):
    """Route based on sentiment"""
    if state["sentiment"] == "negative":
        return "handle_negative"
    elif state["sentiment"] == "positive":
        return "handle_positive"
    else:
        return "handle_neutral"

# Add conditional edges
workflow.add_conditional_edges(
    "sentiment",
    route_by_sentiment,
    {
        "handle_negative": "negative_handler",
        "handle_positive": "positive_handler",
        "handle_neutral": "neutral_handler"
    }
)
```

## Detailed Guides

### [LangChain Patterns](resources/langchain-patterns.md)

- Chain composition
- Memory and context
- Tools and agents
- RAG implementation

### [LangGraph Workflows](resources/langgraph-workflows.md)

- State machines
- Conditional routing
- Multi-agent systems
- Complex orchestration

### [Prompt Engineering](resources/prompt-engineering.md)

- Effective prompt design
- Few-shot learning
- Chain-of-thought prompting
- Prompt templates

### [Model Deployment](resources/model-deployment.md)

- Local model serving
- API integration
- Optimization and caching
- Cost management

### [Testing AI Systems](resources/testing-ai.md)

- Unit testing with mocks
- Integration testing
- Prompt testing
- Evaluation metrics

---

## Quick Reference

### Common LangChain Imports

```python
from langchain_openai import ChatOpenAI, OpenAI
from langchain_anthropic import ChatAnthropic
from langchain.prompts import ChatPromptTemplate, PromptTemplate
from langchain.output_parsers import PydanticOutputParser, StrOutputParser
from langchain.chains import LLMChain, SequentialChain
from langchain.memory import ConversationBufferMemory
from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain.tools import Tool
from langchain.callbacks import get_openai_callback
```

### Common LangGraph Imports

```python
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from operator import add
```

### FastAPI + Streaming

```python
from fastapi.responses import StreamingResponse

async def stream_response():
    async for chunk in chain.astream(input):
        yield f"data: {chunk}\n\n"

return StreamingResponse(stream_response(), media_type="text/event-stream")
```

---

## Resources

- [LangChain Docs](https://python.langchain.com/)
- [LangGraph Docs](https://langchain-ai.github.io/langgraph/)
- [OpenAI Cookbook](https://cookbook.openai.com/)
- [Anthropic Docs](https://docs.anthropic.com/)
- [Prompt Engineering Guide](https://www.promptingguide.ai/)

---

**Remember:** AI features require careful prompt design, error handling, and monitoring. Always validate LLM outputs, implement retry logic, and track costs!