企业官网建设流程全解析

写这套东西的起因很直接：我自己把一套对话问答后端从 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 依然是优秀的框架。管理后台、内容系统、标准业务 CRUD，Django 的 ORM、Admin、中间件体系、迁移机制，到现在依然能打，而且效率极高。

但 AI Agent 后端的画像完全是另一回事。它是一个无状态接口集合，要支持流式返回、长连接、频繁增删工具、返回结构不稳定、还可能对接多家模型供应商。这种服务你不需要 Admin 后台，不需要复杂的 ORM 映射，也不需要表单体系，你更需要的是轻量、异步、类型安全、能轻松写 SSE 流式的框架。

我把两个框架在我心里的体感差别直接列成表格：

这里不是踩一捧一。我的真实建议是：如果你只是要给内部做个带管理后台的 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( model=settings.model_name, api_key=settings.openai_api_key, base_url=settings.openai_base_url, temperature=settings.temperature, max_tokens=settings.max_tokens, timeout=60, )

请求和响应模型用 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_model=ChatResponse) 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(answer=answer, session_id=req.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( agent=agent, tools=tools, verbose=True, handle_parsing_errors=True, max_iterations=5, )

在接口层，我们建议用astream而不是ainvoke，因为 Agent 执行过程会产出多个事件——思考、工具调用、工具结果、最终回答——这些事件可以直接传给前端展示。方便即时观察“Agent 现在在做什么”。

LangChain 目前有两套 Agent 相关方案：早期的是这里的create_react_agent+AgentExecutor，封装完整、上手快；新的 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 实现 SSE

SSE（Server-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 f"data: {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 f"data: {json.dumps({'type': 'tool_call', 'tool': action, 'result': str(result)})}\n\n" elif "output" in event: yield f"data: {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_type="text/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_size=500, chunk_overlap=80, separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?"], ) chunks = text_splitter.split_documents(documents) # 向量化（这里用 Ollama 本地 embedding） embeddings = OllamaEmbeddings(model="nomic-embed-text") # 存入 Chroma vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings, persist_directory="./chroma_db", ) def search_knowledge_base(keyword: str, top_k: int = 3) -> str: """检索知识库，返回最相关的段落拼接结果。""" docs = vectorstore.similarity_search(keyword, k=top_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( retry=retry_if_exception_type(TimeoutError), stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10) ) 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 不是不能用，只是当你发现自己在给它打各种异步补丁的时候，就该考虑换个更顺手的工具了。愿你少踩我踩过的那些坑。