LangChain Python Tutorial: Complete Guide 2025

git clone https://github.com/jetthoughts/langchain-python-examples.git
cd langchain-python-examples
pip install -r requirements.txt
python examples/04_qa_bot.py

python --version
# Should show: Python 3.8.0 or higher (3.11+ recommended)

# Create virtual environment
python -m venv langchain_env

# Activate virtual environment
# On macOS/Linux:
source langchain_env/bin/activate
# On Windows:
langchain_env\Scripts\activate

# Verify activation (should show path to virtual env)
which python

# Core LangChain installation
pip install langchain==0.1.0

# OpenAI integration (most popular)
pip install langchain-openai==0.0.2

# Optional: Anthropic integration for Claude
pip install langchain-anthropic==0.0.2

# Optional: Community integrations
pip install langchain-community==0.0.10

pip install python-dotenv==1.0.0

# .env
OPENAI_API_KEY=sk-your-openai-key-here
ANTHROPIC_API_KEY=sk-ant-your-anthropic-key-here

# .gitignore
.env
.env.local
venv/
langchain_env/
__pycache__/
*.pyc

# main.py
from dotenv import load_dotenv
import os

# Load environment variables from .env file
load_dotenv()

# Access API keys securely
openai_key = os.getenv('OPENAI_API_KEY')
anthropic_key = os.getenv('ANTHROPIC_API_KEY')

# configure_openai.py
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os

load_dotenv()

# Initialize OpenAI LLM
llm = ChatOpenAI(
    model="gpt-4-turbo-preview",
    temperature=0.7,  # 0 = deterministic, 1 = creative
    max_tokens=1000,
    api_key=os.getenv('OPENAI_API_KEY')
)

# Test the connection
response = llm.invoke("Say 'Hello from OpenAI!' in exactly 5 words.")
print(response.content)

# configure_anthropic.py
from langchain_anthropic import ChatAnthropic
from dotenv import load_dotenv
import os

load_dotenv()

# Claude is often better for production (lower hallucination rates)
llm = ChatAnthropic(
    model="claude-3-sonnet-20240229",
    max_tokens=1024,
    temperature=0.5,
    api_key=os.getenv('ANTHROPIC_API_KEY')
)

# Test the connection
response = llm.invoke("Say 'Hello from Claude!' in exactly 5 words.")
print(response.content)

# test_langchain.py
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage
from dotenv import load_dotenv
import os

load_dotenv()

# Initialize LLM
llm = ChatOpenAI(
    model="gpt-4-turbo-preview",
    api_key=os.getenv('OPENAI_API_KEY')
)

# Create messages with system context
messages = [
    SystemMessage(content="You are a helpful Python programming assistant."),
    HumanMessage(content="Explain Python in exactly 3 sentences. Be encouraging!")
]

# Make your first AI call
response = llm.invoke(messages)
print(response.content)

python test_langchain.py

# python_assistant.py
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain.prompts import PromptTemplate
from dotenv import load_dotenv
import os

load_dotenv()

class PythonAssistant:
    def __init__(self):
        # Set up LLM
        self.llm = ChatOpenAI(
            model="gpt-4-turbo-preview",
            temperature=0.7,
            api_key=os.getenv('OPENAI_API_KEY')
        )

        # Set up memory to remember conversation context
        self.memory = ConversationBufferMemory(
            return_messages=True,
            memory_key="chat_history"
        )

        # Create custom prompt template
        prompt = PromptTemplate(
            input_variables=["chat_history", "input"],
            template="""You are a helpful Python programming expert.
            Provide clear, practical advice with code examples.
            Be encouraging and reference Python best practices.

            Conversation history: {chat_history}

            Human: {input}
            Assistant:"""
        )

        # Create conversation chain
        self.conversation = ConversationChain(
            llm=self.llm,
            memory=self.memory,
            prompt=prompt,
            verbose=True  # Shows chain execution steps
        )

    def ask(self, question):
        """Ask a question and get AI response"""
        try:
            response = self.conversation.predict(input=question)
            return response
        except Exception as e:
            return f"Error: {str(e)}"

    def get_conversation_history(self):
        """Get conversation history"""
        return self.memory.load_memory_variables({})

# Usage example
if __name__ == "__main__":
    assistant = PythonAssistant()

    # Ask multiple questions - assistant remembers context!
    print("Q: What's a Python decorator?")
    print(assistant.ask("What's a Python decorator? Give a simple example."))
    print("\n---\n")

    print("Q: How do I use it with classes?")
    # Assistant remembers we're talking about decorators!
    print(assistant.ask("How do I use it with classes?"))
    print("\n---\n")

    # Show conversation history
    print("Conversation so far:")
    history = assistant.get_conversation_history()
    print(history['chat_history'])

from langchain.memory import ConversationBufferMemory

# Keep all conversation history
memory = ConversationBufferMemory(
    return_messages=True,
    memory_key="chat_history"
)

from langchain.memory import ConversationBufferWindowMemory

# Keep last N messages only
memory = ConversationBufferWindowMemory(
    k=10,  # Keep last 10 messages
    return_messages=True,
    memory_key="chat_history"
)

from langchain.memory import ConversationSummaryMemory
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-3.5-turbo")  # Use cheaper model for summaries

# Automatically summarize older messages
memory = ConversationSummaryMemory(
    llm=llm,
    return_messages=True,
    memory_key="chat_history"
)

from langchain.memory import ConversationTokenBufferMemory
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4-turbo-preview")

# Ensure we never exceed token limits
memory = ConversationTokenBufferMemory(
    llm=llm,
    max_token_limit=4000,  # For GPT-4 Turbo (8K context)
    return_messages=True,
    memory_key="chat_history"
)

# calculator_agent.py
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.tools import StructuredTool
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from dotenv import load_dotenv
import os

load_dotenv()

def calculator_tool(expression: str) -> str:
    """
    Evaluate mathematical expressions safely.

    Args:
        expression: Mathematical expression to evaluate (e.g., "2 + 2", "10 * 5")

    Returns:
        Result of the calculation as a string
    """
    try:
        # Safe evaluation (restricted to math operations)
        result = eval(expression, {"__builtins__": {}}, {})
        return str(result)
    except Exception as e:
        return f"Error: {str(e)}"

class CalculatorAgent:
    def __init__(self):
        # Initialize LLM
        self.llm = ChatOpenAI(
            model="gpt-4-turbo-preview",
            api_key=os.getenv('OPENAI_API_KEY')
        )

        # Define calculator tool
        calculator = StructuredTool.from_function(
            func=calculator_tool,
            name="calculator",
            description="Useful for performing mathematical calculations. Input should be a valid mathematical expression."
        )

        # Create agent prompt
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a helpful assistant with access to a calculator. Always show your work."),
            MessagesPlaceholder(variable_name="chat_history", optional=True),
            ("human", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad")
        ])

        # Create agent with tools
        agent = create_openai_functions_agent(
            llm=self.llm,
            tools=[calculator],
            prompt=prompt
        )

        # Create agent executor
        self.agent_executor = AgentExecutor(
            agent=agent,
            tools=[calculator],
            verbose=True
        )

    def calculate(self, problem):
        """Solve a mathematical problem"""
        response = self.agent_executor.invoke({"input": problem})
        return response['output']

# Usage
if __name__ == "__main__":
    calc_agent = CalculatorAgent()

    result = calc_agent.calculate("What's 1,234 * 567 + 89?")
    print(result)
    # Agent will:
    # 1. Recognize it needs to calculate
    # 2. Use the calculator tool: (1234 * 567) + 89
    # 3. Return: "The answer is 699,667"

pip install django>=4.2.0
pip install celery redis
pip install bleach  # For input sanitization (XSS prevention)

# ai_assistant/models.py
from django.db import models
from django.contrib.auth.models import User

class Conversation(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='conversations')
    title = models.CharField(max_length=200)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    status = models.CharField(
        max_length=20,
        choices=[
            ('active', 'Active'),
            ('processing', 'Processing'),
            ('completed', 'Completed'),
            ('error', 'Error')
        ],
        default='active'
    )

    class Meta:
        ordering = ['-updated_at']

    def __str__(self):
        return f"{self.user.username} - {self.title}"

class AIMessage(models.Model):
    conversation = models.ForeignKey(Conversation, on_delete=models.CASCADE, related_name='messages')
    role = models.CharField(
        max_length=20,
        choices=[
            ('user', 'User'),
            ('assistant', 'Assistant'),
            ('system', 'System')
        ]
    )
    content = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True)
    token_count = models.IntegerField(null=True, blank=True)

    class Meta:
        ordering = ['created_at']

    def __str__(self):
        return f"{self.role}: {self.content[:50]}"

class TokenUsage(models.Model):
    conversation = models.ForeignKey(Conversation, on_delete=models.CASCADE, related_name='token_usage')
    prompt_tokens = models.IntegerField()
    completion_tokens = models.IntegerField()
    total_tokens = models.IntegerField()
    estimated_cost_usd = models.DecimalField(max_digits=10, decimal_places=6)
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return f"Conversation {self.conversation.id}: {self.total_tokens} tokens (${self.estimated_cost_usd})"

# ai_assistant/views.py
from django.http import JsonResponse
from django.views.decorators.http import require_http_methods
from django.contrib.auth.decorators import login_required
from django.views.decorators.csrf import csrf_exempt
import json
import bleach  # For input sanitization
from .models import Conversation, AIMessage
from .tasks import generate_ai_response_task

@login_required
@require_http_methods(["POST"])
@csrf_exempt  # Use proper CSRF protection in production
def chat(request):
    """Handle AI chat requests"""
    try:
        data = json.loads(request.body)
        user_message = data.get('message')
        conversation_id = data.get('conversation_id')

        if not user_message:
            return JsonResponse({'error': 'Message is required'}, status=400)

        # Sanitize user input to prevent XSS and prompt injection
        # Allow only safe HTML tags, strip everything else
        user_message = bleach.clean(
            user_message,
            tags=[],  # No HTML tags allowed
            strip=True  # Strip disallowed tags instead of escaping
        ).strip()

        # Additional validation after sanitization
        if not user_message:
            return JsonResponse({'error': 'Message cannot be empty after sanitization'}, status=400)

        # Rate limiting (prevent abuse)
        user_messages_today = AIMessage.objects.filter(
            conversation__user=request.user,
            role='user',
            created_at__date=timezone.now().date()
        ).count()

        if user_messages_today >= 50:
            return JsonResponse({
                'error': 'Daily AI message limit reached (50/day)'
            }, status=429)

        # Get or create conversation
        if conversation_id:
            conversation = Conversation.objects.get(
                id=conversation_id,
                user=request.user
            )
        else:
            conversation = Conversation.objects.create(
                user=request.user,
                title=user_message[:50]  # Use first message as title
            )

        # Store user's message
        AIMessage.objects.create(
            conversation=conversation,
            role='user',
            content=user_message
        )

        # Process with AI asynchronously (recommended for production)
        generate_ai_response_task.delay(
            conversation_id=conversation.id,
            user_message=user_message
        )

        return JsonResponse({
            'status': 'processing',
            'conversation_id': conversation.id,
            'message': 'AI is thinking...'
        }, status=202)

    except Conversation.DoesNotExist:
        return JsonResponse({'error': 'Conversation not found'}, status=404)
    except Exception as e:
        return JsonResponse({'error': str(e)}, status=500)

@login_required
@require_http_methods(["GET"])
def get_conversation(request, conversation_id):
    """Get conversation history"""
    try:
        conversation = Conversation.objects.get(
            id=conversation_id,
            user=request.user
        )

        messages = conversation.messages.all().values('role', 'content', 'created_at')

        return JsonResponse({
            'conversation_id': conversation.id,
            'title': conversation.title,
            'status': conversation.status,
            'messages': list(messages)
        })

    except Conversation.DoesNotExist:
        return JsonResponse({'error': 'Conversation not found'}, status=404)

# ai_assistant/tasks.py
from celery import shared_task
from .models import Conversation, AIMessage, TokenUsage
from .services import AiAssistantService
import logging

logger = logging.getLogger(__name__)

@shared_task(bind=True, max_retries=3)
def generate_ai_response_task(self, conversation_id, user_message):
    """
    Generate AI response asynchronously.

    Args:
        conversation_id: ID of the conversation
        user_message: User's message content
    """
    try:
        conversation = Conversation.objects.get(id=conversation_id)
        conversation.status = 'processing'
        conversation.save()

        # Generate AI response using LangChain service
        service = AiAssistantService(conversation)
        ai_response, usage = service.respond_to(user_message)

        # Store AI response
        AIMessage.objects.create(
            conversation=conversation,
            role='assistant',
            content=ai_response,
            token_count=usage.get('completion_tokens')
        )

        # Track token usage for billing
        TokenUsage.objects.create(
            conversation=conversation,
            prompt_tokens=usage.get('prompt_tokens', 0),
            completion_tokens=usage.get('completion_tokens', 0),
            total_tokens=usage.get('total_tokens', 0),
            estimated_cost_usd=calculate_cost(usage)
        )

        # Update conversation status
        conversation.status = 'completed'
        conversation.save()

        logger.info(f"AI response generated for conversation {conversation_id}")

    except Conversation.DoesNotExist:
        logger.error(f"Conversation {conversation_id} not found")
        raise
    except Exception as e:
        logger.error(f"Error generating AI response: {str(e)}")
        # Retry with exponential backoff
        raise self.retry(exc=e, countdown=2 ** self.request.retries)

def calculate_cost(usage):
    """Calculate estimated cost in USD"""
    # GPT-4 Turbo pricing (as of 2025)
    prompt_cost = usage.get('prompt_tokens', 0) * 0.00001    # $0.01 per 1K tokens
    completion_cost = usage.get('completion_tokens', 0) * 0.00003  # $0.03 per 1K tokens
    return prompt_cost + completion_cost

# ai_assistant/services.py
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from django.core.cache import cache
from django.conf import settings
import hashlib
import os

class AiAssistantService:
    def __init__(self, conversation):
        self.conversation = conversation
        self.llm = ChatOpenAI(
            model="gpt-4-turbo-preview",
            temperature=0.7,
            api_key=os.getenv('OPENAI_API_KEY')
        )

    def respond_to(self, message):
        """Generate AI response with caching"""
        # Cache identical questions (exact match)
        cache_key = f"ai_response:{hashlib.sha256(message.encode()).hexdigest()}"

        cached_response = cache.get(cache_key)
        if cached_response:
            return cached_response, {'prompt_tokens': 0, 'completion_tokens': 0, 'total_tokens': 0}

        # Generate response only if not cached
        response, usage = self._generate_response(message)

        # Cache for 24 hours
        cache.set(cache_key, response, timeout=60*60*24)

        return response, usage

    def _generate_response(self, message):
        """Generate response using LangChain"""
        # Load conversation history for context
        memory = self._build_memory_from_conversation()

        conversation = ConversationChain(
            llm=self.llm,
            memory=memory,
            verbose=settings.DEBUG
        )

        try:
            response = conversation.predict(input=message)

            # Extract token usage (if available)
            usage = {
                'prompt_tokens': getattr(self.llm, 'prompt_tokens', 0),
                'completion_tokens': getattr(self.llm, 'completion_tokens', 0),
                'total_tokens': getattr(self.llm, 'total_tokens', 0)
            }

            return response, usage

        except Exception as e:
            raise Exception(f"AI generation error: {str(e)}")

    def _build_memory_from_conversation(self):
        """Build LangChain memory from Django conversation history"""
        memory = ConversationBufferMemory(
            return_messages=True,
            memory_key="chat_history"
        )

        # Load last 20 messages from database
        messages = self.conversation.messages.all()[:20]
        for msg in messages:
            if msg.role == 'user':
                memory.chat_memory.add_user_message(msg.content)
            elif msg.role == 'assistant':
                memory.chat_memory.add_ai_message(msg.content)

        return memory

# app/main.py
from fastapi import FastAPI, HTTPException, Depends
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from langchain_openai import ChatOpenAI
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from langchain.schema import HumanMessage
from typing import AsyncIterator
import os
from dotenv import load_dotenv

load_dotenv()

app = FastAPI(title="LangChain FastAPI Demo")

class ChatRequest(BaseModel):
    message: str
    stream: bool = False

class ChatResponse(BaseModel):
    response: str
    tokens_used: int

# Initialize LLM with streaming support
llm = ChatOpenAI(
    model="gpt-4-turbo-preview",
    temperature=0.7,
    streaming=True,
    api_key=os.getenv('OPENAI_API_KEY')
)

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    """
    Non-streaming chat endpoint.

    Example request:
    ```
    POST /chat
    {
        "message": "Explain async Python in 2 sentences",
        "stream": false
    }
    ```
    """
    try:
        response = llm.invoke([HumanMessage(content=request.message)])

        return ChatResponse(
            response=response.content,
            tokens_used=response.response_metadata.get('token_usage', {}).get('total_tokens', 0)
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    """
    Streaming chat endpoint with Server-Sent Events.

    Example request:
    ```
    POST /chat/stream
    {
        "message": "Write a Python function to calculate Fibonacci"
    }
    ```

    Response streams token-by-token for real-time UX.
    """
    async def generate() -> AsyncIterator[str]:
        try:
            async for chunk in llm.astream([HumanMessage(content=request.message)]):
                if chunk.content:
                    yield f"data: {chunk.content}\n\n"
            yield "data: [DONE]\n\n"
        except Exception as e:
            yield f"data: Error: {str(e)}\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no"  # Disable nginx buffering
        }
    )

@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {"status": "healthy", "service": "langchain-fastapi"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# app/tasks.py
from celery import Celery
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage
import os

celery_app = Celery(
    'fastapi_ai',
    broker=os.getenv('REDIS_URL', 'redis://localhost:6379/0'),
    backend=os.getenv('REDIS_URL', 'redis://localhost:6379/0')
)

@celery_app.task
def process_ai_request(message: str, user_id: str):
    """Process AI request in background"""
    llm = ChatOpenAI(
        model="gpt-4-turbo-preview",
        api_key=os.getenv('OPENAI_API_KEY')
    )

    response = llm.invoke([HumanMessage(content=message)])

    # Store result in database or cache
    # ... (implementation depends on your storage layer)

    return {
        'user_id': user_id,
        'response': response.content,
        'status': 'completed'
    }

# app/main.py (add background task endpoint)
from app.tasks import process_ai_request as ai_task

@app.post("/chat/async")
async def chat_async(request: ChatRequest, background_tasks: BackgroundTasks):
    """
    Submit AI request for background processing.
    Returns task ID immediately, process result asynchronously.
    """
    task = ai_task.delay(request.message, user_id="example_user")

    return {
        "task_id": task.id,
        "status": "processing",
        "message": "Your request is being processed"
    }

@app.get("/chat/async/{task_id}")
async def get_task_status(task_id: str):
    """Check status of background AI task"""
    task = ai_task.AsyncResult(task_id)

    if task.ready():
        return {
            "task_id": task_id,
            "status": "completed",
            "result": task.result
        }
    else:
        return {
            "task_id": task_id,
            "status": "processing"
        }

# app/dependencies.py
from fastapi import HTTPException, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from langchain.schema import LLMResult
from typing import Optional
import logging

logger = logging.getLogger(__name__)

# Rate limiter
limiter = Limiter(key_func=get_remote_address)

# app/main.py (add rate limiting)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.post("/chat")
@limiter.limit("10/minute")  # 10 requests per minute per IP
async def chat(request: Request, chat_request: ChatRequest):
    """Rate-limited chat endpoint"""
    try:
        response = llm.invoke([HumanMessage(content=chat_request.message)])
        return ChatResponse(
            response=response.content,
            tokens_used=response.response_metadata.get('token_usage', {}).get('total_tokens', 0)
        )
    except Exception as e:
        logger.error(f"Chat error: {str(e)}")
        raise HTTPException(status_code=500, detail="AI service unavailable")

# model_selector.py
from langchain_openai import ChatOpenAI

class ModelSelector:
    @staticmethod
    def select_for_task(task_complexity: str) -> ChatOpenAI:
        """Select most cost-effective model for task"""
        if task_complexity == "simple":
            # Grammar fixing, simple classification
            return ChatOpenAI(model="gpt-3.5-turbo", max_tokens=500)
        elif task_complexity == "moderate":
            # Summaries, basic reasoning
            return ChatOpenAI(model="gpt-4-turbo-preview", max_tokens=1000)
        elif task_complexity == "complex":
            # Advanced reasoning, code generation
            return ChatOpenAI(model="gpt-4", max_tokens=2000)
        else:
            # Default to balanced option
            return ChatOpenAI(model="gpt-4-turbo-preview", max_tokens=1000)

# Usage
llm = ModelSelector.select_for_task("simple")  # Use GPT-3.5-Turbo for simple tasks

# semantic_cache.py
from langchain.cache import RedisCache
from langchain.globals import set_llm_cache
import redis

# Set up Redis cache for LangChain
redis_client = redis.Redis(host='localhost', port=6379, db=0)
set_llm_cache(RedisCache(redis_client))

# Now all LLM calls are automatically cached!
# Identical inputs return cached results instantly (0 cost)

# token_limiter.py
from langchain_openai import ChatOpenAI
from langchain.callbacks import get_openai_callback

def generate_with_budget(prompt: str, max_cost_usd: float = 0.10):
    """Generate response with strict budget limit"""
    llm = ChatOpenAI(
        model="gpt-4-turbo-preview",
        max_tokens=1500  # Hard limit
    )

    with get_openai_callback() as cb:
        response = llm.invoke(prompt)

        # Check if we exceeded budget
        if cb.total_cost > max_cost_usd:
            raise Exception(f"Cost ${cb.total_cost:.4f} exceeded budget ${max_cost_usd}")

        return response.content, cb.total_cost

# Usage
try:
    response, cost = generate_with_budget("Explain quantum physics", max_cost_usd=0.05)
    print(f"Response: {response}")
    print(f"Cost: ${cost:.4f}")
except Exception as e:
    print(f"Budget exceeded: {e}")

# streaming_example.py
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage

llm = ChatOpenAI(model="gpt-4-turbo-preview", streaming=True)

# Stream response token-by-token
for chunk in llm.stream([HumanMessage(content="Write a Python function")]):
    print(chunk.content, end="", flush=True)

# parallel_processing.py
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage
import asyncio

llm = ChatOpenAI(model="gpt-4-turbo-preview")

async def process_batch(messages: list[str]):
    """Process multiple AI requests concurrently"""
    tasks = [
        llm.ainvoke([HumanMessage(content=msg)])
        for msg in messages
    ]

    # Wait for all to complete
    responses = await asyncio.gather(*tasks)
    return [r.content for r in responses]

# Usage
messages = [
    "Summarize Python decorators",
    "Explain async/await",
    "What is FastAPI?"
]

responses = asyncio.run(process_batch(messages))
for i, response in enumerate(responses):
    print(f"Response {i+1}: {response}\n")

# robust_ai_service.py
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain.schema import HumanMessage
import logging

logger = logging.getLogger(__name__)

class RobustAiService:
    def __init__(self):
        # Primary provider: OpenAI
        self.primary_llm = ChatOpenAI(
            model="gpt-4-turbo-preview",
            api_key=os.getenv('OPENAI_API_KEY')
        )

        # Fallback provider: Anthropic Claude
        self.fallback_llm = ChatAnthropic(
            model="claude-3-sonnet-20240229",
            api_key=os.getenv('ANTHROPIC_API_KEY')
        )

    def generate_with_fallback(self, message: str) -> str:
        """Generate response with automatic fallback"""
        # Try primary AI provider
        try:
            response = self.primary_llm.invoke([HumanMessage(content=message)])
            return response.content
        except Exception as e:
            logger.warning(f"Primary AI failed: {str(e)}, falling back to Anthropic")

            # Fallback to secondary provider
            try:
                response = self.fallback_llm.invoke([HumanMessage(content=message)])
                return response.content
            except Exception as e:
                logger.error(f"All AI providers failed: {str(e)}")

                # Ultimate fallback: return helpful error message
                return (
                    "I'm temporarily unavailable. Our team has been notified. "
                    "Please try again in a few minutes."
                )

# Usage
service = RobustAiService()
response = service.generate_with_fallback("Explain Python async")
print(response)

# monitoring.py
from langchain.callbacks.base import BaseCallbackHandler
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage
import logging
import time

logger = logging.getLogger(__name__)

class MonitoringCallback(BaseCallbackHandler):
    """Custom callback for monitoring LangChain operations"""

    def __init__(self):
        self.start_time = None
        self.tokens_used = 0
        self.cost = 0

    def on_llm_start(self, serialized, prompts, **kwargs):
        """Called when LLM starts"""
        self.start_time = time.time()
        logger.info(f"LLM started with {len(prompts)} prompts")

    def on_llm_end(self, response, **kwargs):
        """Called when LLM completes"""
        duration = time.time() - self.start_time

        # Extract token usage
        if hasattr(response, 'llm_output') and response.llm_output:
            token_usage = response.llm_output.get('token_usage', {})
            self.tokens_used = token_usage.get('total_tokens', 0)

            # Calculate cost (GPT-4 Turbo pricing)
            prompt_tokens = token_usage.get('prompt_tokens', 0)
            completion_tokens = token_usage.get('completion_tokens', 0)
            self.cost = (prompt_tokens * 0.00001) + (completion_tokens * 0.00003)

        logger.info(f"LLM completed in {duration:.2f}s, {self.tokens_used} tokens, ${self.cost:.4f}")

        # Alert if costs spike
        if self.cost > 0.50:
            logger.warning(f"High-cost AI request: ${self.cost:.4f}")

# Usage
llm = ChatOpenAI(
    model="gpt-4-turbo-preview",
    callbacks=[MonitoringCallback()]
)

response = llm.invoke([HumanMessage(content="Explain Python async in detail")])
print(response.content)

# security.py
import re
import os
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage

class SecureAiService:
    def __init__(self):
        # API Key Management - NEVER commit keys
        # Use environment variables
        self.llm = ChatOpenAI(
            model="gpt-4-turbo-preview",
            api_key=os.getenv('OPENAI_API_KEY')
        )

    def sanitize_user_input(self, input_text: str) -> str:
        """
        Prevent prompt injection attacks.

        Common injection patterns:
        - "Ignore previous instructions"
        - "You are now..."
        - "Forget everything and..."
        """
        # Remove potential prompt injection attempts
        cleaned = re.sub(r'ignore previous instructions', '', input_text, flags=re.IGNORECASE)
        cleaned = re.sub(r'you are now', '', cleaned, flags=re.IGNORECASE)
        cleaned = re.sub(r'forget everything', '', cleaned, flags=re.IGNORECASE)

        # Limit length to prevent token exhaustion
        return cleaned[:500]

    def filter_ai_response(self, response: str) -> str:
        """Filter inappropriate content from AI responses"""
        # Check for inappropriate content (implement based on your needs)
        if self._contains_inappropriate_content(response):
            logger.warning("AI generated inappropriate content")
            return "I can't provide that information."

        return response

    def _contains_inappropriate_content(self, text: str) -> bool:
        """Check for inappropriate content (simplified example)"""
        # Implement your content filtering logic
        # Consider using moderation APIs (OpenAI Moderation API)
        banned_words = ['offensive_word1', 'offensive_word2']  # Example
        return any(word in text.lower() for word in banned_words)

    def generate_safe_response(self, user_input: str) -> str:
        """Generate AI response with security filters"""
        # Sanitize input
        clean_input = self.sanitize_user_input(user_input)

        # Generate response
        response = self.llm.invoke([HumanMessage(content=clean_input)])

        # Filter output
        safe_response = self.filter_ai_response(response.content)

        return safe_response

pip install pytest pytest-asyncio pytest-mock pytest-cov

mkdir tests
touch tests/__init__.py
touch tests/test_qa_bot.py

"""
Unit tests for Question-Answering Bot
Demonstrates mocking LangChain LLM calls
"""
import pytest
from unittest.mock import MagicMock, patch
from examples.qa_bot import PythonAssistant

class TestPythonAssistant:
    """Test suite for PythonAssistant class"""

    @pytest.fixture
    def mock_llm_response(self):
        """Fixture for mocked LLM response"""
        return "A decorator is a function that modifies another function..."

    @patch('examples.qa_bot.ChatOpenAI')
    def test_assistant_initialization(self, mock_chatgpt):
        """Test that PythonAssistant initializes correctly"""
        # Arrange
        mock_llm = MagicMock()
        mock_chatgpt.return_value = mock_llm

        # Act
        assistant = PythonAssistant()

        # Assert
        assert assistant.llm is not None
        assert assistant.memory is not None
        assert assistant.conversation is not None
        mock_chatgpt.assert_called_once()

    @patch('examples.qa_bot.ChatOpenAI')
    def test_ask_question_success(self, mock_chatgpt, mock_llm_response):
        """Test successful question-answer interaction"""
        # Arrange
        mock_llm = MagicMock()
        mock_chatgpt.return_value = mock_llm

        # Mock the conversation chain's predict method
        mock_predict = MagicMock(return_value=mock_llm_response)

        assistant = PythonAssistant()
        assistant.conversation.predict = mock_predict

        # Act
        question = "What is a Python decorator?"
        response = assistant.ask(question)

        # Assert
        assert response == mock_llm_response
        mock_predict.assert_called_once_with(input=question)
        assert "decorator" in response.lower()

    @patch('examples.qa_bot.ChatOpenAI')
    def test_ask_question_with_context(self, mock_chatgpt):
        """Test that assistant remembers conversation context"""
        # Arrange
        mock_llm = MagicMock()
        mock_chatgpt.return_value = mock_llm

        assistant = PythonAssistant()

        # Mock responses for two questions
        responses = [
            "A decorator is a function...",
            "To use decorators with classes, use @property..."
        ]

        mock_predict = MagicMock(side_effect=responses)
        assistant.conversation.predict = mock_predict

        # Act
        response1 = assistant.ask("What is a Python decorator?")
        response2 = assistant.ask("How do I use it with classes?")

        # Assert
        assert "decorator" in response1.lower()
        assert "class" in response2.lower()
        assert mock_predict.call_count == 2

    @patch('examples.qa_bot.ChatOpenAI')
    def test_ask_question_error_handling(self, mock_chatgpt):
        """Test error handling when LLM call fails"""
        # Arrange
        mock_llm = MagicMock()
        mock_chatgpt.return_value = mock_llm

        assistant = PythonAssistant()

        # Mock exception during prediction
        mock_predict = MagicMock(
            side_effect=Exception("API connection failed")
        )
        assistant.conversation.predict = mock_predict

        # Act
        response = assistant.ask("What is Python?")

        # Assert
        assert "Error" in response
        assert "API connection failed" in response

import pytest
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage

@pytest.mark.asyncio
async def test_async_streaming():
    """Test async streaming responses"""
    # Mock async LLM
    llm = ChatOpenAI(model="gpt-4-turbo-preview")

    # Mock astream method
    async def mock_astream(messages):
        """Mock async stream generator"""
        for chunk in ["Hello", " ", "World"]:
            yield type('Chunk', (), {'content': chunk})()

    llm.astream = mock_astream

    # Collect streamed chunks
    chunks = []
    async for chunk in llm.astream([HumanMessage(content="test")]):
        chunks.append(chunk.content)

    assert chunks == ["Hello", " ", "World"]

import pytest
import os

class TestPythonAssistantIntegration:
    """Integration tests (require actual API keys)"""

    @pytest.mark.integration
    @pytest.mark.skipif(
        not os.getenv('OPENAI_API_KEY'),
        reason="Requires OPENAI_API_KEY environment variable"
    )
    def test_real_api_call(self):
        """Test with real OpenAI API (integration test)"""
        assistant = PythonAssistant()

        question = "What is 2 + 2? Answer with just the number."
        response = assistant.ask(question)

        assert response is not None
        assert len(response) > 0
        # Note: We don't assert exact response since AI output varies

# Run unit tests only (fast, no API calls)
pytest tests/

# Run integration tests (slow, requires API keys)
pytest tests/ -m integration

# Skip integration tests
pytest tests/ -m "not integration"

# Generate coverage report
pytest tests/ --cov=examples --cov-report=html

# View coverage in browser
open htmlcov/index.html

# Set minimum coverage threshold
pytest tests/ --cov=examples --cov-fail-under=90

import pytest
from django.test import Client
from unittest.mock import patch, MagicMock

@pytest.mark.django_db
def test_chat_endpoint():
    """Test Django chat endpoint"""
    client = Client()

    # Mock Celery task
    with patch('myapp.tasks.process_llm_request.delay') as mock_task:
        mock_task.return_value.id = 'test-task-id'

        response = client.post(
            '/api/chat/',
            {'message': 'What is Python?'},
            content_type='application/json'
        )

        assert response.status_code == 202
        assert response.json()['task_id'] == 'test-task-id'
        mock_task.assert_called_once()

from fastapi.testclient import TestClient
from main import app
from unittest.mock import patch, AsyncMock

client = TestClient(app)

def test_fastapi_chat_endpoint():
    """Test FastAPI chat endpoint"""
    with patch('main.llm') as mock_llm:
        # Mock async LLM response
        mock_llm.ainvoke = AsyncMock(
            return_value=type('Response', (), {'content': 'Test response'})()
        )

        response = client.post(
            "/chat",
            json={"message": "Hello"}
        )

        assert response.status_code == 200
        assert "Test response" in response.json()["response"]

@pytest.fixture
def sample_questions():
    """Sample questions for testing"""
    return [
        "What is a Python decorator?",
        "How do I use async/await?",
        "Explain list comprehensions"
    ]

@pytest.fixture
def mock_llm():
    """Mock LLM for testing"""
    with patch('langchain_openai.ChatOpenAI') as mock:
        yield mock

name: Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install pytest pytest-cov pytest-asyncio
      - name: Run tests
        run: pytest tests/ --cov=examples --cov-report=xml
      - name: Upload coverage
        uses: codecov/codecov-action@v3

# 1. Check API key is set correctly
import os
print(os.getenv('OPENAI_API_KEY'))  # Should not be None or empty

# 2. Verify network connectivity
import requests
try:
    response = requests.get('https://api.openai.com/v1/models', timeout=5)
    print(f"API reachable: {response.status_code}")
except Exception as e:
    print(f"Connection error: {e}")

# 3. Check for proxy settings
# If behind corporate firewall, set proxy
os.environ['HTTP_PROXY'] = 'http://your-proxy:8080'
os.environ['HTTPS_PROXY'] = 'http://your-proxy:8080'

# 1. Implement exponential backoff
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage
import time

def complete_with_retry(prompt: str, max_attempts: int = 3):
    """Complete with exponential backoff retry"""
    llm = ChatOpenAI(model="gpt-4-turbo-preview")

    for attempt in range(max_attempts):
        try:
            response = llm.invoke([HumanMessage(content=prompt)])
            return response.content
        except Exception as e:
            if 'rate limit' in str(e).lower() and attempt < max_attempts - 1:
                sleep_time = 2 ** attempt  # Exponential backoff
                print(f"Rate limited, retrying in {sleep_time}s...")
                time.sleep(sleep_time)
            else:
                raise e

# 2. Use Celery/queue systems to throttle requests
# See Django/FastAPI integration sections above

# 1. Use token-aware memory
from langchain.memory import ConversationTokenBufferMemory
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4-turbo-preview")

memory = ConversationTokenBufferMemory(
    llm=llm,
    max_token_limit=4000,  # Stay well under 8K limit
    return_messages=True
)

# 2. Truncate long inputs
import tiktoken

def count_tokens(text: str, model: str = "gpt-4") -> int:
    """Count tokens in text"""
    encoding = tiktoken.encoding_for_model(model)
    return len(encoding.encode(text))

def truncate_to_tokens(text: str, max_tokens: int = 3000) -> str:
    """Truncate text to maximum token count"""
    encoding = tiktoken.encoding_for_model("gpt-4")
    tokens = encoding.encode(text)

    if len(tokens) > max_tokens:
        truncated_tokens = tokens[:max_tokens]
        return encoding.decode(truncated_tokens)

    return text

# Usage
long_text = "..." * 10000
safe_text = truncate_to_tokens(long_text, max_tokens=3000)

# 1. Verify installation
pip show langchain langchain-openai
# Should show installed versions

# 2. Check Python version compatibility
python --version
# Needs Python 3.8+ (3.11+ recommended)

# 3. Update packages and resolve conflicts
pip install --upgrade langchain langchain-openai langchain-community

# 4. Use virtual environment (recommended)
python -m venv venv
source venv/bin/activate  # On macOS/Linux
pip install langchain langchain-openai

# 5. Check for version conflicts
pip list | grep langchain