Django AI 集成：完整指南

Django 团队在添加 AI 功能时，在写任何代码之前都会面临一个架构事实：Django 的请求周期传统上是同步的，而 LLM 调用需要数秒。 每个集成决策都源于你将等待放在哪里。三种可行的放置方式：Celery 任务（Django 惯用的默认方式）、流式传输令牌的异步视图（用于聊天体验），或者——仅用于快速操作——内联在请求中。本指南涵盖所有三种方式，以及如何在现有 Postgres 上使用 pgvector 进行 RAG。

放置方式 1：Celery——Django 处理 AI 任务的答案

不需要实时令牌流的 LLM 工作（保存时总结、分类提交、生成草稿）应放入你可能已经在运行的任务队列中：

python
tasks.py
from celery import shared_task
from openai import OpenAI
@shared_task(bind=True, max_retries=3, retry_backoff=True)
def summarize_article(self, article_id):
    from .models import Article
    article = Article.objects.get(pk=article_id)
    try:
        resp = OpenAI().chat.completions.create(
            model='gpt-4o-mini',
            messages=[{'role': 'user', 'content': f'Summarize in 3 bullets:\n{article.body}'}],
        )
        article.summary = resp.choices[0].message.content
        article.save(update_fields=['summary'])
    except Exception as exc:
        raise self.retry(exc=exc)   # 退避处理速率限制

python
views.py — 请求立即返回；UI 显示“生成中…”并轮询或接收推送
def request_summary(request, pk):
    summarize_article.delay(pk)
    return JsonResponse({'status': 'queued'})

这使 gunicorn 工作进程保持空闲（一个同步工作进程在 OpenAI 上阻塞 8 秒意味着它无法处理请求——在负载下这会导致服务中断），免费提供带退避的重试机制，并留下审计轨迹。除非你特别需要流式传输，否则默认使用此方式。

放置方式 2：用于聊天体验的异步视图 + SSE

Django 多年来一直支持异步视图；在 ASGI 服务器（uvicorn/daphne）下，你可以在没有 Celery 的情况下流式传输令牌：

python
views.py
import json
from django.http import StreamingHttpResponse
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def chat_stream(request):
    prompt = json.loads(request.body)['prompt']
    async def gen():
        stream = await client.chat.completions.create(
            model='gpt-4o-mini',
            messages=[{'role': 'user', 'content': prompt}],
            stream=True,
        )
        async for chunk in stream:
            if token := (chunk.choices[0].delta.content or ''):
                yield f'data: {json.dumps({"token": token})}\n\n'
        yield 'data: [DONE]\n\n'
    resp = StreamingHttpResponse(gen(), content_type='text/event-stream')
    resp['Cache-Control'] = 'no-cache'
    resp['X-Accel-Buffering'] = 'no'   # nginx：不要缓冲
    return resp

容易出问题的要求：在 ASGI 下运行（uvicorn project.asgi:application——通过 WSGI 流式传输会为每个连接的客户端阻塞一个工作进程），使用 AsyncOpenAI（在异步视图中使用同步客户端会阻塞事件循环——参见 同步与异步 LLM 调用对比），并且不要用 @login_required 风格的同步中间件包装生成器，这会强制同步模式。异步代码中的 ORM 调用需要使用 await Model.objects.aget(...) 或 sync_to_async。客户端的 fetch/parse 代码与 FastAPI 流式传输配方 相同。

在你已有的 Postgres 上做 RAG：pgvector

Django 团队很少需要一开始就使用专门的向量数据库——pgvector 将你现有的 Postgres 变成一个向量数据库，通过 django-pgvector 提供迁移和 ORM 集成：

python
models.py
from pgvector.django import VectorField, HnswIndex
class DocChunk(models.Model):
    document = models.ForeignKey('Document', on_delete=models.CASCADE)
    text = models.TextField()
    embedding = VectorField(dimensions=1536)   # 匹配你的嵌入模型
    class Meta:
        indexes = [HnswIndex(name='chunk_emb_idx', fields=['embedding'],
                             opclasses=['vector_cosine_ops'])]
查询
from pgvector.django import CosineDistance
hits = (DocChunk.objects
        .annotate(dist=CosineDistance('embedding', query_embedding))
        .order_by('dist')[:5])

嵌入生成应放在 Celery 中（分块→嵌入→保存，批量处理）。一个事务性数据库，一个备份方案，SQL 过滤器可以与向量搜索组合。扩展上限以及何时升级到专用存储：参见 pgvector 指南。

Django 特定的生产注意事项

设置规范：通过 environ/密钥管理器管理密钥，绝不要写在 settings.py 字面量中；在设置中固定模型名称，以便提示/模型更改是配置变更，而非代码部署。

成本防护：对 AI 端点实施每用户速率限制（django-ratelimit），并将每次请求的令牌使用量记录到你的数据库中——你有 ORM，用它来构建审计表。

将管理后台作为 AI 运维控制台：注册审计/任务模型——Django 管理后台是一个免费的面板，用于检查失败的生成和每个用户的支出。

缓存重复生成：使用框架缓存，键为（模型 + 模板版本 + 输入）的哈希值——推荐使用 Redis 后端。

FAQ

我应该为 AI 部分切换到 FastAPI 吗？ 如果 AI 接口是一个小的流式 API，其余部分是经典的 Django，那么一个 sidecar FastAPI 服务是干净的——但异步 Django 视图现在已基本弥补了差距；不要仅仅为了一个端点而拆分整个技术栈。

Channels/WebSockets？ 仅用于双向需求（协作会话、语音）。单向令牌流是 SSE 的领域——参见 流式传输与轮询对比。

在 Django 中使用 LangChain？ 在它发挥价值的地方（复杂的 RAG/智能体——LangChain vs LlamaIndex）是可以的；对于简单的调用，使用提供商 SDK 加 Celery 更轻量。

---

*最后更新：2026 年 6 月。*