Python AI 开发栈 2026：FastAPI + LangChain + Supabase

现代 Python AI 栈结合了 FastAPI（异步 API）、LangChain（LLM 编排）和 Supabase（数据库 + 向量存储）。

项目设置

bash
mkdir ai-backend && cd ai-backend
python -m venv venv && source venv/bin/activate
pip install fastapi uvicorn langchain langchain-openai langchain-anthropic \
    supabase python-dotenv pydantic tiktoken python-multipart

流式聊天端点

python
from fastapi import APIRouter
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from typing import List
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
router = APIRouter()
class ChatMessage(BaseModel):
    role: str
    content: str
class ChatRequest(BaseModel):
    messages: List[ChatMessage]
    system_prompt: str = 'You are a helpful assistant.'
llm = ChatAnthropic(model='claude-sonnet-4-5', streaming=True)
@router.post('/stream')
async def chat_stream(request: ChatRequest):
    async def generate():
        messages = [SystemMessage(content=request.system_prompt)]
        for msg in request.messages:
            if msg.role == 'user':
                messages.append(HumanMessage(content=msg.content))
            else:
                messages.append(AIMessage(content=msg.content))
        
        async for chunk in llm.astream(messages):
            if chunk.content:
                yield f'data: {chunk.content}\n\n'
        yield 'data: [DONE]\n\n'
    
    return StreamingResponse(generate(), media_type='text/event-stream')

文档上传 + RAG

python
from fastapi import UploadFile, File, HTTPException
from supabase import create_client
from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader
import tempfile, os
supabase = create_client(SUPABASE_URL, SUPABASE_KEY)
embeddings = OpenAIEmbeddings(model='text-embedding-3-small')
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
@router.post('/upload')
async def upload_document(file: UploadFile = File(...)):
    with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as tmp:
        tmp.write(await file.read())
        tmp_path = tmp.name
    
    try:
        loader = PyPDFLoader(tmp_path)
        chunks = splitter.split_documents(loader.load())
        texts = [c.page_content for c in chunks]
        chunk_embeddings = embeddings.embed_documents(texts)
        
        records = [
            {'content': texts[i], 'embedding': chunk_embeddings[i],
             'metadata': {'source': file.filename}}
            for i in range(len(chunks))
        ]
        supabase.table('document_chunks').insert(records).execute()
        return {'message': f'Processed {len(chunks)} chunks'}
    finally:
        os.unlink(tmp_path)
@router.post('/query')
async def rag_query(question: str, top_k: int = 5):
    q_embedding = embeddings.embed_query(question)
    
    result = supabase.rpc(
        'match_documents',
        {'query_embedding': q_embedding, 'match_count': top_k}
    ).execute()
    
    context = '\n\n'.join([r['content'] for r in result.data])
    
    llm = ChatAnthropic(model='claude-sonnet-4-5')
    response = llm.invoke(
        f'Answer based only on context:\n{context}\n\nQuestion: {question}'
    )
    
    return {'answer': response.content, 'sources': result.data}

Supabase 向量设置

sql
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE document_chunks (
    id UUID DEFAULT gen_random_uuid() PRIMARY KEY,
    content TEXT NOT NULL,
    embedding vector(1536),
    metadata JSONB DEFAULT '{}',
    created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX ON document_chunks USING ivfflat (embedding vector_cosine_ops)
    WITH (lists = 100);
CREATE OR REPLACE FUNCTION match_documents(
    query_embedding vector(1536),
    match_count INT DEFAULT 5
)
RETURNS TABLE(id UUID, content TEXT, metadata JSONB, similarity FLOAT)
LANGUAGE SQL AS $$
    SELECT id, content, metadata,
           1 - (embedding <=> query_embedding) AS similarity
    FROM document_chunks
    ORDER BY embedding <=> query_embedding
    LIMIT match_count;
$$;

使用 Docker 部署

dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

yaml
docker-compose.yml
version: '3.8'
services:
  api:
    build: .
    ports:
      - '8000:8000'
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - SUPABASE_URL=${SUPABASE_URL}
      - SUPABASE_SERVICE_KEY=${SUPABASE_SERVICE_KEY}

结论

FastAPI + LangChain + Supabase 是 2026 年生产级 Python AI 栈。FastAPI 完美处理异步流式传输，LangChain 编排复杂的 LLM 工作流，Supabase 在一个托管平台中同时提供数据库和向量搜索。这消除了大多数应用对独立向量数据库的需求。