写这套东西的起因很直接我自己把一套对话问答后端从 Django 迁到了 FastAPI。当时接的是一个基于大模型的项目线上用户一多Django 那边线程池先被打满接着是流式输出怎么也推不动前端只能靠轮询假装“打字机”。折腾完那一次之后再有人跟我提“用 Django 写 AI Agent 后端”我都只剩一句话别真的别。不是 Django 不好是这个场景它确实不对味。这篇文章就是那次迁移和之后重写服务的经验整理。你会看到为什么 LLM 接口和 Agent 服务天然适合 FastAPI而不是 Django也会看到我从零搭起来的一个可运行的 FastAPI LangChain 工程包括目录结构、普通对话接口、Agent 工具循环、SSE 流式输出、本地知识库以及生产环境里那些躲不开的细节。适合正在写 LLM 服务、想搭 AI Agent 练手项目、或者准备 FastAPI 和 LangChain 相关面试的人。1. 为什么是 FastAPI 而不是 Django先说一个很容易被忽略的事实LLM 接口的本质不是普通 HTTP 请求而是“长任务异步化”。1.1 LLM 接口的本质是异步长连接你调用一次大模型不是发过去就秒回。从请求进入服务端开始到模型把完整结果生成出来中间往往要几十秒甚至更久。这个过程中服务端和客户端之间的连接必须一直保持而且最好是一边生成、一边把 token 推给前端否则用户就对着空白页面干等。这种场景下Django 默认的 WSGI 模型是致命的。WSGI 是同步阻塞模型一个请求进来就长期占着一个线程。你说 Djangon 还能用 Channels 走 ASGI能但那是打补丁的路线。Django 社区的主流写法、生态习惯、中间件设计全是按同步 WSGI 来的硬改成异步之后ORM 调用、缓存读取、一些第三方库都会变成同步阻塞你依然被卡住。FastAPI 不一样。它底层是 Starlette天生就是异步事件循环。请求发过来之后遇到模型响应这种需要等待的操作就挂起这个协程事件循环立刻腾出手去处理其他请求。同样是 100 个并发连接Django 可能已经打满了线程池FastAPI 在等待期间几乎不占什么资源。这就是两者最本质的差别。1.2 Django 不是不行而是这个场景不对味我先把话放这儿Django 依然是优秀的框架。管理后台、内容系统、标准业务 CRUDDjango 的 ORM、Admin、中间件体系、迁移机制到现在依然能打而且效率极高。但 AI Agent 后端的画像完全是另一回事。它是一个无状态接口集合要支持流式返回、长连接、频繁增删工具、返回结构不稳定、还可能对接多家模型供应商。这种服务你不需要 Admin 后台不需要复杂的 ORM 映射也不需要表单体系你更需要的是轻量、异步、类型安全、能轻松写 SSE 流式的框架。我把两个框架在我心里的体感差别直接列成表格维度DjangoFastAPI请求模型WSGI 同步每请求占一线程ASGI 异步事件循环复用LLM 长等待线程被白白占住协程挂起几乎不耗资源流式输出SSE需要 Channels 全套基础设施原生 StreamingResponse 就搞定返回结构校验DRF Serializer偏重Pydantic 模型原生支持工具调用的动态路由需要自己拼适配层类型即契约天然适合参数校验生态节奏大而全起步慢小而精起步快这里不是踩一捧一。我的真实建议是如果你只是要给内部做个带管理后台的 AI 业务系统Django 完全没问题。但如果你做的是面向 Web、App、小程序多端复用的 LLM 接口服务或者一个 Agent 网关那就没必要背 Django 那套重型全家桶FastAPI 是更顺手的起点。1.3 还有一个很多人忽略的优势类型契约做 LLM 接口最痛苦的事情不是模型不聪明而是前端不知道你会返回什么。今天的接口返回{ answer: 你好 }明天你为了加引用来源就变成{ answer: 你好, sources: [] }后天又要加工具调用记录又变成{ message: {...}, tool_calls: [...] }。前端每改一次接口就要联调一次吵一次架。FastAPI 加 Pydantic 解决的就是这件事。你定义一个AgentEvent模型字段、类型、默认值全都写在代码里自动生成 OpenAPI 文档前端照着文档写代码。这一点在 Agent 场景下尤其重要——因为 Agent 的返回不是一个字符串而是一连串事件流没有类型契约前端根本不知道该怎么拼装。Django 里要做同类事情得上 DRF Serializer而且大量团队实际是直接手写 dict 返回。这样小项目能忍一上 Agent 就崩。2. 从零搭出你的第一个 LLM 接口方向确定了直接进入实操。我用一个最小可跑的工程带你走完 FastAPI LangChain 的完整链路。2.1 项目目录结构怎么组织最清晰FastAPI 的项目结构没有官方标准但基于我跑了多个项目的经验LLM 服务建议这样分层llm-service/ ├── app/ │ ├── main.py # FastAPI 实例入口 │ ├── api/ │ │ └── v1/ │ │ ├── __init__.py │ │ ├── chat.py # 对话接口 │ │ └── agent.py # Agent 接口 │ ├── core/ │ │ ├── __init__.py │ │ ├── config.py # 全局配置读 .env │ │ └── llm_client.py # 模型客户端统一包装 │ ├── schemas/ │ │ ├── __init__.py │ │ ├── common.py # 通用响应模型 │ │ └── chat.py # 对话请求/响应模型 │ ├── services/ │ │ ├── __init__.py │ │ ├── agent_engine.py # Agent 编排逻辑 │ │ └── rag.py # 知识库检索服务 │ └── tools/ │ ├── __init__.py │ └── registry.py # Agent 工具注册表 ├── tests/ ├── pyproject.toml └── .env.example各层职责很清晰api只做路由和参数校验不写业务逻辑services承接对话、Agent、知识库这样的核心编排tools专门放 Agent 能调用的工具core是底层配置和客户端单例。这样分层的核心原因只有一个AI 项目变化太快。今天你用的模型是 A明天可能就要换 B今天是一个 Agent明天就是三个 Agent 协作。如果所有东西揉在路由里换模型的时候就要动接口层的代码动静太大。2.2 最小可跑代码/api/v1/chat接口我们从对话接口写起。先写配置文件用pydantic-settings读环境变量这样密钥永远不会进代码仓库# app/core/config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): model_config {env_file: .env, env_file_encoding: utf-8} model_name: str gpt-4o-mini openai_api_key: str openai_base_url: str https://api.openai.com/v1 temperature: float 0.7 max_tokens: int 2048 settings Settings()然后是模型客户端的包装。这里我建议直接用 LangChain 的ChatOpenAI因为它天然兼容各类 OpenAI 风格接口后面接 Agent 也方便# app/core/llm_client.py from langchain_openai import ChatOpenAI from core.config import settings llm ChatOpenAI( modelsettings.model_name, api_keysettings.openai_api_key, base_urlsettings.openai_base_url, temperaturesettings.temperature, max_tokenssettings.max_tokens, timeout60, )请求和响应模型用 Pydantic 定死# app/schemas/chat.py from pydantic import BaseModel, Field class ChatRequest(BaseModel): message: str Field(..., description用户消息) session_id: str Field(default, description会话 ID用于区分上下文) system_prompt: str Field(, description可选的系统提示词) class ChatResponse(BaseModel): answer: str session_id: str路由层不做任何业务逻辑校验通过后直接调 service# app/api/v1/chat.py from fastapi import APIRouter from langchain_core.messages import HumanMessage from langchain_core.output_parsers import StrOutputParser from core.llm_client import llm from schemas.chat import ChatRequest, ChatResponse router APIRouter(prefix/api/v1/chat, tags[chat]) router.post(, response_modelChatResponse) async def chat(req: ChatRequest) - ChatResponse: chain llm | StrOutputParser() answer await chain.ainvoke( [{role: system, content: req.system_prompt or 你是一个有用的助手}, {role: user, content: req.message}] ) return ChatResponse(answeranswer, session_idreq.session_id)这样最快能跑通。但注意session_id在这里还没真正用于上下文记忆只是预留。真正做多轮对话时你需要一个会话存储层把历史消息按session_id存起来拼进消息列表。我后续的 Agent 章节会展示更完整的做法。2.3 为什么直接用 LangChain而不是裸调 SDK有读者可能会问既然 ChatOpenAI 自己都能调为什么中间套一层 LangChain我的看法如果只做一个最简单的对话接口裸调 SDK 完全没问题。但只要你的需求开始走向 Agent、知识库、多模型切换LangChain 的抽象价值就体现出来了。它定义了一套统一的BaseChatModel接口今天你接 OpenAI明天换本地 Ollama后天换 Azure OpenAI只需要改一行配置。工具调用、输出解析、记忆管理LangChain 都有现成的组件。当然LangChain 也不是银弹。它的抽象层有时会让你觉得绕调试链路很长。我的建议是项目初期先裸调 SDK 跑通逻辑然后逐渐引入 LangChain 的模型封装和工具框架不要一上来就全家桶。3. 从普通接口进阶到 AI Agent对话接口只是热身。标题里说的 AI Agent才是这段的核心。3.1 Agent 和普通接口的本质区别普通接口是你把用户输入拼进 prompt丢给模型拿到一次回答。它是“一次调用一次返回”。Agent 是“循环决策”。模型拿到问题后先判断自己需不需要外部信息。如果需要就吐出一个工具调用请求——这时候你不能直接把结果返回给用户而是要把这个工具调用请求取出来真正去执行工具拿到结果再把它塞回给模型。模型看到工具结果后可能给出最终回答也可能再次发起新的调用。就这样 Reason → Act → Observe → Reason 循环直到模型觉得信息够了。3.2 把工具定义清楚docstring 就是模型的说明书在 LangChain 里定义工具最简单的方式就是用tool装饰器# app/tools/registry.py from datetime import datetime from langchain_core.tools import tool tool def get_current_time() - str: 获取当前系统时间的函数。当用户询问现在几点、今天日期时使用。 return datetime.now().strftime(%Y-%m-%d %H:%M:%S) tool def query_knowledge_base(keyword: str) - str: 在内部知识库中检索资料返回与关键字最相关的片段。 当用户询问公司制度、产品功能、技术文档时优先使用。 from services.rag import search_knowledge_base return search_knowledge_base(keyword)这里有个很多人忽略的细节函数名和 docstring 极其关键。模型不是看你函数里面的代码来决定调不调用而是看你的函数签名和 docstring。描述写得模糊比如“获取信息”模型就不知道什么时候该调描述写得具体比如“用户询问产品功能时优先调用”模型的工具选择准确率会明显上升。我见过不少团队在 Agent 上线后频繁报provider rejected the request schema or tool payload错误排查到最后发现是工具函数的参数 schema 和实际调用对不上。比如模型想传keyword你定义的参数名却是query模型推理工具参数时就会产生非法 payload。所以工具函数的参数名、类型、默认值一定要和 docstring 里描述的保持一致这个稳定性比模型本身的聪明程度更重要。3.3 组装 Agent 并流式消费事件用 LangChain 的create_react_agent加AgentExecutor是最快的 Agent 实现方式# app/services/agent_engine.py from langchain import hub from langchain.agents import AgentExecutor, create_react_agent from core.llm_client import llm from tools.registry import get_current_time, query_knowledge_base tools [get_current_time, query_knowledge_base] prompt hub.pull(hwchase17/react) agent create_react_agent(llm, tools, prompt) agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue, max_iterations5, )在接口层我们建议用astream而不是ainvoke因为 Agent 执行过程会产出多个事件——思考、工具调用、工具结果、最终回答——这些事件可以直接传给前端展示。方便即时观察“Agent 现在在做什么”。LangChain 目前有两套 Agent 相关方案早期的是这里的create_react_agentAgentExecutor封装完整、上手快新的 LangGraph 则是更底层的图编排框架节点、状态、边全由自己定义灵活但复杂不少。面试官常拿这两个做对比我的建议是先学会 AgentExecutor 的用法理解 ReAct 循环是怎么回事再碰 LangGraph。直接上手 LangGraph 很容易糊成一团。3.4 状态、记忆与工具设计的“三元组”视角做 Agent 设计时我常用一个来自搜索引擎原理的三元组框架来整理系统提示词和工具描述key我是谁定义 Agent 的角色和记忆基础。比如“你是某电商平台的智能客服你清楚平台的退货政策”。这个部分决定模型以什么立场回答问题。query我在找什么定义当前任务的意图。它决定 Agent 要不要调工具、调哪个工具、以及站在什么目标上做检索。你在 prompt 中写明“当你需要价格信息时必须查询产品数据库”就是在强化这个 query 层。value我能提供什么工具的输出、知识库检索到的资料实际回答问题的素材。这一层决定最终回答的质量。把这三层拆开写比把一个巨大的系统提示词糊在一起要好得多。因为 Agent 的每个工具调用本质上都是一次“意图 → 检索 → 供给”的循环你需要给模型明确的边界。4. 流式输出与实时交互给 LLM 接口做流式输出不是加分项是基本体验。4.1 没有流式输出用户根本等不住一个正常的模型回答可能需要 30 到 40 秒。如果这 40 秒内用户什么都看不到他大概率会觉得服务挂了。但如果你让第一个 token 在 2 秒内出现然后像打字机一样逐字刷新用户感知的等待时间就会大幅度缩短。我在 Django 版本的后端里卡得最久的就是这件事。当时用的是 WSGI 部署响应被 uWSGI 缓冲前端拿不到增量内容只有请求结束后一下子全拿到流式形同虚设。最后被迫改成短轮询体验别提多差了。换到 FastAPI问题迎刃而解。4.2 用 StreamingResponse 实现 SSESSEServer-Sent Events是实现流式效果最标准的协议。它本质上是 HTTP 长连接服务端按data: {json}\n\n格式持续推送数据。FastAPI 里写 SSE 极其简单就是一个异步生成器加一个响应包装# app/api/v1/agent.py import asyncio import json from fastapi import APIRouter from fastapi.responses import StreamingResponse from schemas.chat import ChatRequest from services.agent_engine import agent_executor router APIRouter(prefix/api/v1/agent, tags[agent]) async def event_generator(req: ChatRequest): # 先发一个 session 开始事件让前端知道连接已建立 yield fdata: {json.dumps({type: session_start, session_id: req.session_id})}\n\n async for event in agent_executor.astream( {input: req.message, session_id: req.session_id} ): # AgentExecutor 会产生多种事件统一打包成 SSE 帧 if steps in event: for step in event[steps]: action step[0].tool result step[1] yield fdata: {json.dumps({type: tool_call, tool: action, result: str(result)})}\n\n elif output in event: yield fdata: {json.dumps({type: token, content: event[output]})}\n\n await asyncio.sleep(0.01) # 稍微降速避免过快的刷新让前端卡顿 yield data: [DONE]\n\n router.post(/stream) async def agent_stream(req: ChatRequest): return StreamingResponse( event_generator(req), media_typetext/event-stream, headers{Cache-Control: no-cache, X-Accel-Buffering: no}, )这里有一个关键细节是X-Accel-Buffering: no这个响应头。如果你部署在 Nginx 后面不加这个头Nginx 会默认缓冲响应SSE 照样变成一次性的。这个坑很隐蔽我第一次部署时没加前端还是拿不到增量排查了半天。4.3 前端拿到 SSE 之后怎么拼装前端拿到事件流后按类型处理session_start清空对话区域初始化 session。tool_call展示 Agent 正在调用哪个工具增强透明感。token把content追加到当前回答区域。[DONE]关闭连接。如果你前端用的 fetch注意要这样读流const res await fetch(/api/v1/agent/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ message: 帮我查一下本周会议安排 }), }); const reader res.body.getReader(); const decoder new TextDecoder(); while (true) { const { value, done } await reader.read(); if (done) break; const chunk decoder.decode(value); // 按 \n\n 拆分成帧再解析 data: 后面的 JSON handleChunk(chunk); }这一整套在 FastAPI 里是标准操作放到 Django 里你需要先解决 Channels、Redis Channel Layer、ASGI 配置、同步转异步一堆问题工作量完全不在一个量级。5. 知识库增强与检索标题里的 LLM 接口如果只对接大模型本身很多场景都没法落地。原因很简单大模型不知道你公司内部的产品文档、不掌握实时数据也没法保证每次都回答准确。知识库RAG就是解决这个问题的。5.1 RAG 的核心链路RAG 的本质是“检索增强生成”不直接让模型瞎编而是先从知识库里检索出相关片段再把片段和用户问题一起交给模型生成。离线阶段把文档切块、向量化、存入向量数据库。在线阶段把用户问题向量化在向量库里做相似度检索取 TopK 片段拼进 prompt。LangChain 对这条链路封装得很成熟。一个本地知识库的搭建用ollama做模型推理、langchain做链路编排、chroma做向量存储三件套就能跑通。5.2 一份可直接跑的本地知识库代码# app/services/rag.py from langchain_chroma import Chroma from langchain_community.document_loaders import DirectoryLoader from langchain_community.embeddings import OllamaEmbeddings from langchain_text_splitters import RecursiveCharacterTextSplitter # 文档加载按扩展名过滤只加载 md 和 txt 文件 loader DirectoryLoader(./docs, glob**/*.{md,txt}) documents loader.load() # 文本切分按语义块切分避免把一个完整主题切成碎渣 text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap80, separators[\n\n, \n, 。, , , ., !, ?], ) chunks text_splitter.split_documents(documents) # 向量化这里用 Ollama 本地 embedding embeddings OllamaEmbeddings(modelnomic-embed-text) # 存入 Chroma vectorstore Chroma.from_documents( documentschunks, embeddingembeddings, persist_directory./chroma_db, ) def search_knowledge_base(keyword: str, top_k: int 3) - str: 检索知识库返回最相关的段落拼接结果。 docs vectorstore.similarity_search(keyword, ktop_k) return \n\n.join([doc.page_content for doc in docs])注意RecursiveCharacterTextSplitter的separators参数。它是按优先级递归切分的先按段落、再按换行、再按句号。中文文本如果没有中文标点的分隔符配置切分效果会很差经常把语义割断。这一点极其影响检索质量。实际项目里你不需要回答每个问题都走一遍 RAG。更合理的做法是把我上面写的query_knowledge_base注册成 Agent 的一个工具让模型自己判断遇到需要事实性资料的问题就调用检索问寒暄话就直接回答。这样既准确又省钱省时间。5.3 向量检索的局限和常见坑很多人第一次做 RAG 都以为向量检索万能真实体验会发现效果不稳定。常见坑有两个第一embedding 模型和业务领域不匹配。通用 embedding 模型对专业领域的术语理解不佳比如医疗、法律、代码。有条件的话用领域数据微调 embedding或者至少多测几个模型对比效果。第二chunk 大小选择随意。切得太大检索结果里噪声多切得太小语义不完整。500 字加 80 字重叠是我的常用起点但具体要看文档类型。产品 FAQ 可以小一点技术文档可以大一点没有银弹要实验。6. 生产落地你躲不开的那些细节最后讲生产环境。很多项目 PoC 跑得通一上线就崩问题几乎都出在这一章。6.1 超时、重试与容错LLM 接口的稳定性比普通 API 差很多上游超时、限流、网络抖动都是常态。所以你的服务必须做好容错。LangChain 的ChatOpenAI可以配合max_retries和timeout参数。但注意重试要区分错误类型429 限流可以退避重试401 认证失败重试没意义。我在生产环境里会用tenacity包做更精细的重试策略或者在服务层自己捕获异常后降级回答。from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential retry( retryretry_if_exception_type(TimeoutError), stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10) ) async def safe_llm_call(messages): return await llm.ainvoke(messages)6.2 并发控制与限流这是一个容易被忽略的问题你的业务接口并发可能很高但大模型本身的吞吐有限你不能让所有请求同时打过去否则上游限流甚至会封你的 key。FastAPI 里最简单的方案是用asyncio.Semaphore控制同时进入模型层的并发数import asyncio llm_semaphore asyncio.Semaphore(10) async def limited_llm_call(messages): async with llm_semaphore: return await llm.ainvoke(messages)到底是 10 个并发合适还是 50 个合适取决于你模型服务的吞吐。如果是云端 API看你的限流配额如果是内网部署的开源模型看显存和推理引擎瓶颈。我一般先压测再根据 p95 延迟倒推并发上限。另外对外接口一定要加一层简单的限流中间件否则一个刷接口的请求就能把你整条链路打满。6.3 LLM 网关统一出口当你的服务开始接多家模型比如成本敏感的场景接开源模型、高质量问答接云端旗舰模型你就会发现“统一出口”变得非常重要。LangChain 的BaseChatModel抽象天然就是网关的角色。你可以在core/llm_client.py里做一个简单的模型路由默认模型成本优先响应快适合普通对话。强化模型效果优先适合复杂推理和 Agent 工具选择。本地模型数据不出内网适合隐私敏感场景。这样改造之后业务代码完全无感只需要在配置里调整路由策略。很多团队现在做所谓“AI Agent 中台”本质上就是这一层网关加工具注册表加权限控制。6.4 密钥与安全密钥永远不要硬编码进代码。.env文件加.gitignore用pydantic-settings注入。对外暴露的服务必须加鉴权哪怕只是一个简单的 Bearer Token 中间件。如果你的服务要接企业内部系统要特别注意工具调用的权限边界——Agent 能调的工具越多权限横切面就越大这是目前 AI 应用最容易出事的地方。注意任何情况下都不要把系统提示词、知识库内容、密钥信息通过流式接口原样返回给前端。Agent 能访问的不等于前端能看到的。最后说几句实际体验从 Django 迁到 FastAPI 之后我最大的感受不是性能数字提升了多少——虽然并发确实上去了——而是整个开发节奏变了。写 SSE 流式输出变得像写普通接口一样自然给 Agent 加新工具只需要新增一个tool函数接口的返回结构被 Pydantic 定死之后前端联调再也没吵过架。如果一定要我给一条建议不要一上来就同时上全套 LangChain。先把 FastAPI 的异步、SSE、Pydantic 这三样吃透再逐步引入 LangChain 的模型封装和 Agent 编排。Django 不是不能用只是当你发现自己在给它打各种异步补丁的时候就该考虑换个更顺手的工具了。愿你少踩我踩过的那些坑。