AI-Powered Django Development: Production-Ready AI Agents Framework

# Traditional approach: Generic LLM calls, no types, no Django integration
response = openai.chat.completions.create(...)  # ❌ No type safety, no Django context
 
# Django-CFG AI Agents: Type-safe, Django-native, production-ready
@content_agent.tool
async def get_user_orders(ctx: RunContext[DjangoDeps]) -> list[Order]:
    user = await ctx.deps.get_user()
    return await Order.objects.filter(user=user).order_by('-created_at')[:10]
 
result = await content_agent.run("Analyze user purchase patterns", deps=deps)
# ✅ Full typing, Django ORM access, automatic caching, cost tracking

# What Copilot suggests (generic Python)
def get_user_data(user_id):
    """Get user data from database"""
    # Copilot suggests generic SQLAlchemy or raw SQL
    with Session() as session:
        result = session.query(User).filter(User.id == user_id).first()
    return result
 
# What you need (Django-specific)
def get_user_data(user_id: int) -> User:
    """Get user data with related objects (see /fundamentals/system/django-integration)"""
    return User.objects.select_related(
        'profile', 'subscription'
    ).prefetch_related(
        'orders__items', 'support_tickets'
    ).get(id=user_id)

# Copilot: Isolated function suggestions
def process_order(order_id):
    # ❌ Just the function logic, no integration
 
# Django-CFG Agents: Complete workflow
class OrderProcessor(Agent[OrderInput, OrderResult]):
    """End-to-end order processing with Django integration"""
 
    @tool
    async def validate_inventory(ctx: RunContext[DjangoDeps], order_id: int) -> bool:
        """Check inventory via Django ORM"""
        order = await Order.objects.aget(id=order_id)
        # Access Django models, check stock, update database
 
    @tool
    async def process_payment(ctx: RunContext[DjangoDeps], order_id: int) -> PaymentResult:
        """Process payment via Stripe integration"""
        # Django-CFG has built-in payment integration
 
    async def run(self, deps: OrderInput) -> OrderResult:
        """Orchestrate: validate → charge → fulfill → notify"""
        # Complete business logic with Django context

from django_cfg.apps.agents import Agent
from pydantic import BaseModel  # Type-safe models (see /fundamentals/core/type-safety)
from typing import TypeVar, Generic
 
# Define input/output types
class ContentInput(BaseModel):
    text: str
    user_id: int
    analyze_sentiment: bool = True
 
class ContentAnalysis(BaseModel):
    sentiment: str  # "positive" | "negative" | "neutral"
    key_topics: list[str]
    summary: str
    confidence_score: float
 
# Create type-safe agent
class ContentAnalyzer(Agent[ContentInput, ContentAnalysis]):
    """Analyze content with Django user context"""
 
    model = "gpt-4o-mini"  # Or claude-3-5-sonnet-20241022
    system_prompt = """
    You are a content analysis expert integrated with Django.
    Analyze content and provide structured insights.
    """
 
    def run(self, deps: ContentInput) -> ContentAnalysis:
        prompt = f"Analyze this content: {deps.text}"
        response = self.call_llm(prompt)
        return ContentAnalysis.model_validate_json(response)

from django_cfg.apps.agents import DjangoDeps
from django.contrib.auth.models import User
from django.http import HttpRequest
 
# Create dependencies from request
async def my_view(request: HttpRequest):
    deps = await DjangoDeps.from_request(request)
 
    # DjangoDeps provides:
    print(deps.user)           # Current Django user
    print(deps.request)        # Original request object
    print(deps.session)        # Django session
    print(deps.permissions)    # User permissions
 
# Tools can access Django ORM
@my_agent.tool
async def get_user_data(ctx: RunContext[DjangoDeps]) -> dict:
    """Access Django models via context"""
    user = await ctx.deps.get_user()  # Async Django ORM
 
    profile = await user.profile.aget()  # Related objects
    orders = await Order.objects.filter(user=user).acount()
 
    return {
        "username": user.username,
        "email": user.email,
        "total_orders": orders,
    }

from django_cfg.apps.agents import agent_registry
from pydantic_ai import RunContext
 
# Register reusable agent
content_agent = agent_registry.create(
    name="content_processor",
    model="gpt-4o-mini",
    system_prompt="You process and analyze content."
)
 
# Tool #1: Database access
@content_agent.tool
async def get_recent_posts(ctx: RunContext[DjangoDeps]) -> list[dict]:
    """Get user's recent blog posts from Django ORM"""
    user = await ctx.deps.get_user()
 
    posts = await BlogPost.objects.filter(
        author=user
    ).order_by('-published_at')[:5].values('title', 'content', 'published_at')
 
    return list(posts)
 
# Tool #2: External API call
@content_agent.tool
async def check_content_moderation(ctx: RunContext[DjangoDeps], text: str) -> bool:
    """Check if content passes moderation via external API"""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.moderation-service.com/check",
            json={"text": text}
        )
    return response.json()["is_safe"]
 
# Tool #3: File operations
@content_agent.tool
async def save_analysis_report(
    ctx: RunContext[DjangoDeps],
    content: str,
    filename: str
) -> str:
    """Save analysis report to Django media storage"""
    from django.core.files.base import ContentFile
    from myapp.models import AnalysisReport
 
    report = await AnalysisReport.objects.acreate(
        user=await ctx.deps.get_user(),
        content=content,
        file=ContentFile(content.encode(), name=filename)
    )
 
    return report.file.url  # Returns media URL

from django_cfg.apps.agents import Agent
from myapp.models import SupportTicket, KnowledgeBaseArticle
from pydantic import BaseModel
 
class TicketInput(BaseModel):
    ticket_id: int
    user_question: str
 
class TicketResponse(BaseModel):
    answer: str
    confidence: float
    escalate: bool
    suggested_articles: list[str]
 
class SupportAgent(Agent[TicketInput, TicketResponse]):
    """Tier-1 customer support automation"""
 
    model = "claude-3-5-sonnet-20241022"
    system_prompt = """
    You are a customer support agent for our Django SaaS platform.
    Answer questions using knowledge base and order history.
    Escalate to human if confidence < 0.8 or issue is complex.
    """
 
    @tool
    async def search_knowledge_base(
        ctx: RunContext[DjangoDeps],
        query: str
    ) -> list[str]:
        """Search knowledge base articles (vector similarity, see /features/built-in-apps/ai-knowledge/overview)"""
        from django.contrib.postgres.search import TrigramSimilarity
 
        articles = await KnowledgeBaseArticle.objects.annotate(
            similarity=TrigramSimilarity('content', query)
        ).filter(similarity__gt=0.3).order_by('-similarity')[:3].values_list('content', flat=True)
 
        return list(articles)
 
    @tool
    async def get_user_orders(ctx: RunContext[DjangoDeps]) -> list[dict]:
        """Get user's recent orders for context"""
        user = await ctx.deps.get_user()
        orders = await Order.objects.filter(user=user).order_by('-created_at')[:5].values(
            'id', 'status', 'total', 'created_at'
        )
        return list(orders)
 
    @tool
    async def create_escalation(
        ctx: RunContext[DjangoDeps],
        reason: str
    ) -> int:
        """Escalate ticket to human agent"""
        ticket = await SupportTicket.objects.aget(id=ctx.deps.ticket_id)
        ticket.status = 'escalated'
        ticket.escalation_reason = reason
        await ticket.asave()
        return ticket.id
 
# Use in view
async def handle_support_ticket(request, ticket_id):
    agent = SupportAgent()
    deps = await DjangoDeps.from_request(request)
 
    result = await agent.run(TicketInput(
        ticket_id=ticket_id,
        user_question=request.POST['question']
    ), deps=deps)
 
    if result.escalate:
        # Human agent notified
        return JsonResponse({"status": "escalated", "message": result.answer})
    else:
        # AI handled it
        return JsonResponse({"status": "resolved", "answer": result.answer})

from django_cfg.apps.agents import Agent
from myapp.models import Product, ProductReview
 
class ProductAnalysisInput(BaseModel):
    product_id: int
    generate_description: bool = True
    analyze_reviews: bool = True
 
class ProductAnalysisOutput(BaseModel):
    seo_title: str
    seo_description: str
    key_features: list[str]
    review_sentiment: float  # -1.0 to 1.0
    common_complaints: list[str]
    recommended_improvements: list[str]
 
class ProductAnalyzer(Agent[ProductAnalysisInput, ProductAnalysisOutput]):
    """Analyze products and generate marketing content"""
 
    model = "gpt-4o"
    system_prompt = """
    You are an e-commerce product analysis expert.
    Generate SEO-optimized content and analyze customer feedback.
    """
 
    @tool
    async def get_product_details(ctx: RunContext[DjangoDeps], product_id: int) -> dict:
        """Get product specifications from Django ORM"""
        product = await Product.objects.select_related('category', 'brand').aget(id=product_id)
 
        return {
            "name": product.name,
            "category": product.category.name,
            "brand": product.brand.name,
            "price": float(product.price),
            "specifications": product.specifications,  # JSONField
        }
 
    @tool
    async def get_reviews(ctx: RunContext[DjangoDeps], product_id: int) -> list[dict]:
        """Get product reviews for sentiment analysis"""
        reviews = await ProductReview.objects.filter(
            product_id=product_id,
            approved=True
        ).order_by('-created_at')[:50].values('rating', 'content', 'created_at')
 
        return list(reviews)
 
# Background task processing (see /features/integrations/django-rq/overview)
import django_rq
 
def analyze_product_async(product_id: int):
    """Process product analysis as background task"""
    agent = ProductAnalyzer()
 
    result = await agent.run(ProductAnalysisInput(
        product_id=product_id,
        generate_description=True,
        analyze_reviews=True
    ))
 
    # Save results to database
    product = await Product.objects.aget(id=product_id)
    product.seo_title = result.seo_title
    product.seo_description = result.seo_description
    product.ai_analysis = result.dict()  # JSONField
    await product.asave()

from django_cfg.apps.agents import Agent
from django.db.models import Sum, Count, Avg
from datetime import datetime, timedelta
 
class ReportInput(BaseModel):
    start_date: datetime
    end_date: datetime
    metrics: list[str]  # ["revenue", "users", "churn"]
 
class ReportOutput(BaseModel):
    executive_summary: str
    key_metrics: dict[str, float]
    trends: list[str]
    recommendations: list[str]
    charts: list[dict]  # Chart data for frontend
 
class BusinessIntelligenceAgent(Agent[ReportInput, ReportOutput]):
    """Generate business intelligence reports"""
 
    model = "claude-3-opus-20240229"  # Most powerful for analysis
    system_prompt = """
    You are a business analyst generating executive reports.
    Analyze data, identify trends, provide actionable recommendations.
    """
 
    @tool
    async def get_revenue_metrics(
        ctx: RunContext[DjangoDeps],
        start: datetime,
        end: datetime
    ) -> dict:
        """Calculate revenue metrics from Django ORM"""
        orders = await Order.objects.filter(
            created_at__gte=start,
            created_at__lte=end,
            status='completed'
        ).aggregate(
            total_revenue=Sum('total'),
            avg_order_value=Avg('total'),
            order_count=Count('id')
        )
        return orders
 
    @tool
    async def get_user_growth(
        ctx: RunContext[DjangoDeps],
        start: datetime,
        end: datetime
    ) -> dict:
        """Calculate user growth metrics"""
        new_users = await User.objects.filter(
            date_joined__gte=start,
            date_joined__lte=end
        ).count()
 
        active_users = await User.objects.filter(
            last_login__gte=start,
            last_login__lte=end
        ).count()
 
        return {
            "new_users": new_users,
            "active_users": active_users,
            "activation_rate": (active_users / new_users * 100) if new_users > 0 else 0
        }
 
    @tool
    async def get_churn_analysis(
        ctx: RunContext[DjangoDeps],
        start: datetime,
        end: datetime
    ) -> dict:
        """Analyze customer churn"""
        churned = await Subscription.objects.filter(
            cancelled_at__gte=start,
            cancelled_at__lte=end
        ).aggregate(
            churned_count=Count('id'),
            churned_revenue=Sum('plan__price')
        )
        return churned
 
# Scheduled task (runs every Monday, see /features/integrations/django-rq/overview for scheduling)
# Configured in config.py with RQScheduleConfig(cron="0 9 * * 1")
def generate_weekly_report():
    """Auto-generate weekly executive report"""
    agent = BusinessIntelligenceAgent()
 
    end_date = datetime.now()
    start_date = end_date - timedelta(days=7)
 
    result = await agent.run(ReportInput(
        start_date=start_date,
        end_date=end_date,
        metrics=["revenue", "users", "churn"]
    ))
 
    # Save report
    report = await ExecutiveReport.objects.acreate(
        title=f"Weekly Report - {start_date.strftime('%Y-%m-%d')}",
        summary=result.executive_summary,
        metrics=result.key_metrics,
        trends=result.trends,
        recommendations=result.recommendations,
        charts=result.charts
    )
 
    # Email to executives
    await send_email(
        to=["[email protected]", "[email protected]"],
        subject=f"Weekly Report - {start_date.strftime('%Y-%m-%d')}",
        template="executive_report.html",
        context={"report": report}
    )

# Django-CFG provides DjangoAgent with Django integration
from pydantic_ai import Agent as PydanticAgent
from django_cfg.apps.agents.core.django_agent import DjangoAgent
 
class DjangoAgent(Generic[DepsT, OutputT]):
    """
    Django-integrated agent using Pydantic AI.
 
    Features:
    - Optimal model selection (GPT-4o-mini via OpenRouter/OpenAI)
    - Automatic fallback to OpenAI models
    - Metrics collection
    - Error handling
    - Tool management
    """
 
    def __init__(
        self,
        name: str,
        deps_type: Type[DepsT],
        output_type: Type[OutputT],
        instructions: str,
        model: Optional[str] = None,
        tools: Optional[Sequence] = None,
        system_prompt: Optional[str] = None,
        **kwargs
    ):
        """Initialize Django Agent with Pydantic AI."""
        # Initialize with automatic model selection and Django integration
        # Automatically selects GPT-4o-mini via OpenRouter (supports tools)
 
    async def run(
        self,
        prompt: str,
        deps: DepsT,
        **kwargs
    ) -> OutputT:
        """Run agent with Django context and automatic metrics tracking"""
 
        # Track execution metrics
        execution = await AgentExecution.objects.acreate(
            agent_name=self.name,
            user=deps.user if hasattr(deps, 'user') else None,
            prompt=prompt,
            status='running'
        )
 
        try:
            # Call parent Pydantic AI run method
            result = await self.run(prompt, deps=deps, **kwargs)
 
            # Update execution record
            execution.status = 'completed'
            execution.output = result.output.dict()
            execution.tokens_used = result.usage.total_tokens
            execution.cost = result.usage.cost
            await execution.asave()
 
            return result
 
        except Exception as e:
            execution.status = 'failed'
            execution.error = str(e)
            await execution.asave()
            raise

# api/config.py
from django_cfg import DjangoConfig
from .environment import env
 
class MyConfig(DjangoConfig):
    # Enable AI agents (auto-configures everything)
    enable_agents: bool = True
 
    # Required: OpenAI API key
    openai_api_key: str = env.openai_api_key
 
    # Optional: Other LLM providers
    anthropic_api_key: str | None = env.anthropic_api_key

# myapp/agents/hello_agent.py
from django_cfg.apps.agents import Agent, agent_registry, DjangoDeps
from pydantic import BaseModel
from pydantic_ai import RunContext
 
class GreetingInput(BaseModel):
    name: str
    language: str = "en"
 
class GreetingOutput(BaseModel):
    message: str
    fun_fact: str
 
# Create agent
hello_agent = agent_registry.create(
    name="hello_world",
    model="gpt-4o-mini",
    system_prompt="You are a friendly assistant that greets users."
)
 
# Add tool for database access
@hello_agent.tool
async def get_user_join_date(ctx: RunContext[DjangoDeps]) -> str:
    """Get when user joined the platform"""
    user = await ctx.deps.get_user()
    return user.date_joined.strftime("%B %d, %Y")

# myapp/views.py
from django.http import JsonResponse
from django_cfg.apps.agents import get_agent, DjangoDeps
from .agents.hello_agent import GreetingInput
 
async def hello_view(request):
    """AI-powered greeting endpoint"""
 
    # Get registered agent
    agent = get_agent("hello_world")
 
    # Create dependencies from request
    deps = await DjangoDeps.from_request(request)
 
    # Run agent
    result = await agent.run(
        prompt=f"Greet {request.user.username} in a friendly way",
        deps=deps
    )
 
    return JsonResponse({
        "message": result.output.message,
        "fun_fact": result.output.fun_fact,
        "cost": result.usage.cost,  # Token cost tracking
    })

from django_cfg.apps.agents import SimpleOrchestrator, Agent
 
# Define agents
email_classifier = Agent(name="email_classifier", ...)
spam_detector = Agent(name="spam_detector", ...)
priority_ranker = Agent(name="priority_ranker", ...)
auto_responder = Agent(name="auto_responder", ...)
 
# Create orchestrator
email_workflow = SimpleOrchestrator(name="email_processing")
 
# Sequential execution
await email_workflow.execute(
    pattern="sequential",
    agents=[
        (email_classifier, {"email": email_data}),
        (spam_detector, {"classification": prev_result}),
        (priority_ranker, {"spam_score": prev_result}),
        (auto_responder, {"priority": prev_result})
    ]
)
 
# Parallel execution
await email_workflow.execute(
    pattern="parallel",
    agents=[
        (sentiment_analyzer, {"text": email_body}),
        (language_detector, {"text": email_body}),
        (category_predictor, {"text": email_body})
    ]
)
 
# Conditional execution
await email_workflow.execute(
    pattern="conditional",
    condition=lambda result: result.spam_score > 0.8,
    if_true_agent=spam_handler,
    if_false_agent=priority_classifier
)

# In Django-CFG settings
class MyConfig(DjangoConfig):
    enable_agents: bool = True
 
    # Configure monitoring
    agent_monitoring: AgentMonitoringConfig = AgentMonitoringConfig(
        alert_on_failures=True,
        alert_threshold=5,  # Alert after 5 failures
        cost_budget_daily=100.00,  # $100/day budget
        alert_on_budget_exceed=True
    )

from django_cfg.apps.agents import Agent
 
class ExpensiveAnalyzer(Agent):
    """Agent with automatic caching"""
 
    # Enable caching (uses Django cache backend)
    enable_cache = True
    cache_ttl = 3600  # 1 hour
 
    async def run(self, prompt: str, **kwargs):
        # Identical prompts return cached results
        # No API call, zero cost
        return await super().run(prompt, **kwargs)
 
# Usage
agent = ExpensiveAnalyzer()
 
# First call: Hits API ($0.002)
result1 = await agent.run("Analyze this content: ...")
 
# Second call (within 1 hour): Returns cached ($0.00)
result2 = await agent.run("Analyze this content: ...")  # Same prompt
 
# After 1 hour: Cache expires, new API call

# All agent executions tracked in database
from django_cfg.apps.agents.models import AgentExecution
 
# Get total cost for last month
from django.db.models import Sum
from datetime import datetime, timedelta
 
total_cost = await AgentExecution.objects.filter(
    created_at__gte=datetime.now() - timedelta(days=30)
).aggregate(total=Sum('cost'))
 
print(f"Total LLM cost last month: ${total_cost['total']:.2f}")
 
# Cost by agent
cost_by_agent = await AgentExecution.objects.values('agent_name').annotate(
    total_cost=Sum('cost'),
    total_tokens=Sum('tokens_used'),
    avg_cost=Avg('cost')
).order_by('-total_cost')

# Automatically pause agents if budget exceeded
if total_cost['total'] > 1000:  # $1000 budget
    # Disable agents temporarily
    MyConfig.enable_agents = False
    # Send alert to admin
    send_alert_email("AI budget exceeded!")