Step-by-Step Guide to Building AI-Powered APIs with FastAPI 0.110 and LangChain 0.3

import logging
import os
from contextlib import asynccontextmanager
from typing import Any, Dict, Optional

from fastapi import FastAPI, HTTPException, Request, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

# Configure module-level logger with structured output for production tracing
logging.basicConfig(
    level=logging.INFO,
    format=\"%(asctime)s - %(name)s - %(levelname)s - %(message)s\"
)
logger = logging.getLogger(__name__)

# Lifespan context manager to handle startup/shutdown events (FastAPI 0.110+ best practice)
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: initialize LangChain components, check API keys, warm caches
    logger.info(\"Starting up AI API service...\")
    yield
    # Shutdown: clean up LLM connections, flush logs, close database pools
    logger.info(\"Shutting down AI API service...\")

# Initialize FastAPI with lifespan, OpenAPI metadata for documentation
app = FastAPI(
    lifespan=lifespan,
    title=\"AI-Powered API\",
    description=\"Production-ready API built with FastAPI 0.110 and LangChain 0.3\",
    version=\"1.0.0\",
    contact={
        \"name\": \"Engineering Team\",
        \"email\": \"eng@example.com\"
    }
)

# CORS middleware configuration for cross-origin requests (adjust origins for production!)
app.add_middleware(
    CORSMiddleware,
    allow_origins=os.getenv(\"ALLOWED_ORIGINS\", \"http://localhost:3000\").split(\",\"),
    allow_credentials=True,
    allow_methods=[\"GET\", \"POST\"],
    allow_headers=[\"*\"],
)

# Custom exception handler for generic 500 errors to avoid leaking sensitive data
@app.exception_handler(Exception)
async def generic_exception_handler(request: Request, exc: Exception):
    logger.error(f\"Unhandled exception: {exc}\", exc_info=True)
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={
            \"error\": \"Internal server error\",
            \"request_id\": request.headers.get(\"X-Request-ID\", \"unknown\")
        }
    )

# Health check endpoint for load balancers and k8s probes
@app.get(\"/health\", tags=[\"Monitoring\"])
async def health_check():
    return {
        \"status\": \"healthy\",
        \"version\": app.version,
        \"fastapi_version\": \"0.110.0\",
        \"langchain_version\": \"0.3.0\"
    }

# Base response model for consistent API output
class BaseResponse(BaseModel):
    success: bool = Field(..., description=\"Whether the request succeeded\")
    data: Optional[Any] = Field(None, description=\"Response payload if successful\")
    error: Optional[str] = Field(None, description=\"Error message if failed\")
    request_id: str = Field(..., description=\"Unique request identifier for tracing\")

if __name__ == \"__main__\":
    import uvicorn
    # Run with 4 workers for production, adjust based on CPU cores
    uvicorn.run(
        \"main:app\",
        host=\"0.0.0.0\",
        port=8000,
        workers=4,
        log_level=\"info\"
    )

import logging
import os
from typing import AsyncGenerator, List, Optional

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_community.llms import OpenAI  # LangChain 0.3 uses langchain_community for legacy LLMs
from langchain_openai import ChatOpenAI  # Preferred for OpenAI-compatible APIs in 0.3
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

# Configure service-level logger
logger = logging.getLogger(__name__)

# Custom exception for LLM-related errors to handle separately from generic API errors
class LLMServiceError(Exception):
    def __init__(self, message: str, status_code: int = 502):
        self.message = message
        self.status_code = status_code
        super().__init__(self.message)

# Retry decorator for transient LLM failures (rate limits, network blips)
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=60),
    retry=retry_if_exception_type((ConnectionError, TimeoutError)),
    reraise=True
)
async def _invoke_llm_with_retry(chain, input: dict):
    \"\"\"Invoke LLM chain with automatic retry for transient failures\"\"\"
    return await chain.ainvoke(input)

class ChatService:
    def __init__(self):
        # Initialize LLM with environment variables, validate on startup
        self.openai_api_key = os.getenv(\"OPENAI_API_KEY\")
        if not self.openai_api_key:
            raise ValueError(\"OPENAI_API_KEY environment variable is required\")

        # LangChain 0.3 ChatOpenAI supports async native, temperature 0 for deterministic outputs
        self.llm = ChatOpenAI(
            model=\"gpt-3.5-turbo\",
            temperature=0,
            api_key=self.openai_api_key,
            max_retries=0  # Handle retries via tenacity for better control
        )

        # Prompt template with system message and chat history placeholder
        self.prompt = ChatPromptTemplate.from_messages([
            SystemMessage(content=\"You are a helpful AI assistant that provides concise, accurate answers.\"),
            MessagesPlaceholder(variable_name=\"chat_history\"),
            HumanMessage(content=\"{user_input}\")
        ])

        # Output parser to extract string from LLM response
        self.output_parser = StrOutputParser()

        # Compose the chain: prompt | llm | output parser
        self.chain = self.prompt | self.llm | self.output_parser
        logger.info(\"ChatService initialized with LangChain 0.3\")

    async def generate_response(
        self,
        user_input: str,
        chat_history: Optional[List[dict]] = None
    ) -> str:
        \"\"\"Generate LLM response with chat history support\"\"\"
        try:
            # Convert chat history dicts to LangChain message objects
            lc_chat_history = []
            if chat_history:
                for msg in chat_history:
                    if msg[\"role\"] == \"user\":
                        lc_chat_history.append(HumanMessage(content=msg[\"content\"]))
                    elif msg[\"role\"] == \"assistant\":
                        lc_chat_history.append(SystemMessage(content=msg[\"content\"]))

            # Invoke chain with retry logic
            response = await _invoke_llm_with_retry(
                self.chain,
                {\"user_input\": user_input, \"chat_history\": lc_chat_history}
            )
            return response
        except Exception as e:
            logger.error(f\"LLM invocation failed: {e}\", exc_info=True)
            raise LLMServiceError(f\"Failed to generate response: {str(e)}\") from e

    async def stream_response(
        self,
        user_input: str,
        chat_history: Optional[List[dict]] = None
    ) -> AsyncGenerator[str, None]:
        \"\"\"Stream LLM response token by token for low-latency UX\"\"\"
        try:
            lc_chat_history = []
            if chat_history:
                for msg in chat_history:
                    if msg[\"role\"] == \"user\":
                        lc_chat_history.append(HumanMessage(content=msg[\"content\"]))
                    elif msg[\"role\"] == \"assistant\":
                        lc_chat_history.append(SystemMessage(content=msg[\"content\"]))

            # Stream tokens from the chain
            async for token in self.chain.astream({
                \"user_input\": user_input,
                \"chat_history\": lc_chat_history
            }):
                yield token
        except Exception as e:
            logger.error(f\"LLM stream failed: {e}\", exc_info=True)
            raise LLMServiceError(f\"Failed to stream response: {str(e)}\") from e

import logging
import hashlib
import json
from typing import List, Optional

from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

from llm_service import ChatService, LLMServiceError

# Configure router and logger
router = APIRouter(prefix=\"/api/v1/chat\", tags=[\"Chat\"])
logger = logging.getLogger(__name__)

# Initialize ChatService as a singleton (use dependency injection in production)
chat_service = ChatService()

# Request model with validation
class ChatRequest(BaseModel):
    user_input: str = Field(..., min_length=1, max_length=4096, description=\"User's input message\")
    chat_history: Optional[List[dict]] = Field(
        default_factory=list,
        description=\"Previous chat history in {role: str, content: str} format\"
    )
    stream: bool = Field(default=False, description=\"Whether to stream the response\")

# Response model for non-streaming requests
class ChatResponse(BaseModel):
    response: str = Field(..., description=\"LLM-generated response\")
    chat_id: str = Field(..., description=\"Unique ID for this chat exchange\")

# In-memory cache for repeat queries (replace with Redis in production)
RESPONSE_CACHE = {}

def get_cache_key(user_input: str, chat_history: List[dict]) -> str:
    \"\"\"Generate deterministic cache key from input and history\"\"\"
    cache_data = {\"user_input\": user_input, \"chat_history\": chat_history}
    return hashlib.sha256(json.dumps(cache_data, sort_keys=True).encode()).hexdigest()

@router.post(\"/completions\", response_model=ChatResponse)
async def create_chat_completion(request: ChatRequest, req: Request):
    \"\"\"Create a non-streaming chat completion\"\"\"
    request_id = req.headers.get(\"X-Request-ID\", \"unknown\")
    logger.info(f\"Processing chat request {request_id}\")

    try:
        # Check cache first to reduce LLM costs
        cache_key = get_cache_key(request.user_input, request.chat_history)
        if cache_key in RESPONSE_CACHE:
            logger.info(f\"Cache hit for request {request_id}\")
            cached = RESPONSE_CACHE[cache_key]
            return ChatResponse(
                response=cached[\"response\"],
                chat_id=cached[\"chat_id\"]
            )

        # Generate response via LangChain service
        response_text = await chat_service.generate_response(
            user_input=request.user_input,
            chat_history=request.chat_history
        )

        # Store in cache (TTL 1 hour in production)
        chat_id = hashlib.md5(f\"{request_id}{request.user_input}\".encode()).hexdigest()
        RESPONSE_CACHE[cache_key] = {
            \"response\": response_text,
            \"chat_id\": chat_id
        }

        return ChatResponse(
            response=response_text,
            chat_id=chat_id
        )
    except LLMServiceError as e:
        logger.error(f\"LLM error for request {request_id}: {e}\")
        raise HTTPException(status_code=e.status_code, detail=e.message)
    except Exception as e:
        logger.error(f\"Unexpected error for request {request_id}: {e}\", exc_info=True)
        raise HTTPException(status_code=500, detail=\"Internal server error\")

@router.post(\"/completions/stream\")
async def create_streaming_chat_completion(request: ChatRequest, req: Request):
    \"\"\"Create a streaming chat completion (returns text/event-stream)\"\"\"
    request_id = req.headers.get(\"X-Request-ID\", \"unknown\")
    logger.info(f\"Processing streaming chat request {request_id}\")

    try:
        # Stream response token by token
        return StreamingResponse(
            chat_service.stream_response(
                user_input=request.user_input,
                chat_history=request.chat_history
            ),
            media_type=\"text/event-stream\"
        )
    except LLMServiceError as e:
        logger.error(f\"LLM stream error for request {request_id}: {e}\")
        raise HTTPException(status_code=e.status_code, detail=e.message)
    except Exception as e:
        logger.error(f\"Unexpected stream error for request {request_id}: {e}\", exc_info=True)
        raise HTTPException(status_code=500, detail=\"Internal server error\")

from pydantic import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate

# Define type-safe output model (Pydantic v2)
class SentimentAnalysis(BaseModel):
    sentiment: str = Field(..., pattern=\"^(positive|negative|neutral)$\")
    confidence: float = Field(..., ge=0.0, le=1.0)
    key_phrases: list[str] = Field(..., min_length=1)

# Initialize parser with model
parser = PydanticOutputParser(pydantic_object=SentimentAnalysis)

# Prompt that instructs LLM to output JSON matching the model
prompt = ChatPromptTemplate.from_messages([
    (\"system\", \"Analyze the sentiment of the user input. {format_instructions}\"),
    (\"human\", \"{user_input}\")
]).partial(format_instructions=parser.get_format_instructions())

# Chain: prompt | llm | parser (returns validated SentimentAnalysis object)
chain = prompt | llm | parser

import tiktoken
from fastapi import Request, Response
from fastapi.middleware.base import BaseHTTPMiddleware

class CostTrackingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Count input tokens from request body (simplified for example)
        body = await request.body()
        encoding = tiktoken.get_encoding(\"cl100k_base\")
        input_tokens = len(encoding.encode(body.decode())) if body else 0

        # Process request
        response = await call_next(request)

        # Get output tokens from response (in production, extract from LLM response)
        output_tokens = 150  # Placeholder: extract from LangChain response

        # Calculate cost (GPT-3.5 Turbo pricing)
        input_cost = (input_tokens / 1000) * 0.0015
        output_cost = (output_tokens / 1000) * 0.002
        total_cost = input_cost + output_cost

        # Log cost for metrics collection
        logger.info(f\"Request cost: ${total_cost:.4f} (input: {input_tokens}, output: {output_tokens})\")
        return response

from langchain_core.callbacks import BaseCallbackHandler
import logging

logger = logging.getLogger(__name__)

class RequestTracingCallback(BaseCallbackHandler):
    def __init__(self, request_id: str):
        self.request_id = request_id
        super().__init__()

    def on_llm_start(self, serialized: dict, prompts: list, **kwargs):
        logger.info(f\"[{self.request_id}] LLM start: {serialized['name']}, prompts: {len(prompts)}\")

    def on_llm_end(self, response, **kwargs):
        logger.info(f\"[{self.request_id}] LLM end: response length {len(response.generations[0][0].text)}\")

    def on_llm_error(self, error, **kwargs):
        logger.error(f\"[{self.request_id}] LLM error: {error}\")

# Pass callback to chain invocation
response = await chain.ainvoke(
    {\"user_input\": \"Hello\"},
    config={\"callbacks\": [RequestTracingCallback(request_id=\"12345\")]}
)

ai-powered-api/
├── main.py                  # Base FastAPI application
├── llm_service.py           # LangChain 0.3 chat service
├── api/
│   ├── __init__.py
│   └── chat.py              # Chat API endpoints
├── requirements.txt         # Pinned dependencies
├── .env.example             # Environment variable template
├── Dockerfile               # Production container image
└── tests/
    ├── __init__.py
    ├── test_main.py         # Health check and exception handler tests
    └── test_llm_service.py  # LangChain integration tests