1. 为什么我劝你别一上来就写代码很多人看到AI全栈项目这几个字第一反应是打开编辑器先pip install一堆包再说。我带过几个刚入行的朋友十个里有八个是这么干的结果无一例外第三天就开始在群里问为什么我的接口返回422为什么向量检索出来的东西驴唇不对马嘴前端跨域又炸了。问题不在于他们技术差而在于没有先想清楚这个系统到底由哪几块拼起来、每块之间的数据是怎么流动的。一个能跑起来、还能撑住真实使用的大模型应用本质上是四层结构叠在一起最底下是模型能力层调用大模型、Embedding、重排往上是编排层LangChain负责把检索、工具调用、对话历史串成一条链再往上是服务层FastAPI把编排逻辑包装成HTTP接口最上面是交互层Vue3把结果渲染成人能看的东西。这四层任何一层没想明白后面都是返工。我这次要拆的这个项目就是把这四层完整地走一遍FastAPI做后端骨架Vue3做前端界面LangChain做编排RAG解决模型不知道你私有数据的问题Agent解决模型不能主动干活的问题Harness解决怎么稳定地测、稳定地跑的问题。听起来名词很多但拆开看每一块都对应一个非常具体的痛点。这篇文章适合谁如果你已经会一点Python、能看懂基本的HTTP请求、对Vue有一点概念哪怕只是改过别人的页面那这篇就是给你写的。如果你是完全零基础也能看但建议先把Python的虚拟环境、pip、基本的函数定义搞明白再回来不然会在环境配置那一步卡很久。我下面不会给你一个复制粘贴就能跑的完整仓库——那种东西网上太多了而且往往跑不起来。我要给你的是每一层为什么这么设计、关键代码长什么样、以及我在实际搭的时候踩过的那些坑。你照着这个思路走最后得到的是一个你自己能改、能扩展、能讲清楚原理的项目而不是一个黑盒。2. 项目骨架四层结构怎么切才不返工2.1 先画数据流再建文件夹我见过太多项目文件夹建得漂漂亮亮models/、services/、utils/分得清清楚楚结果一跑起来发现数据在层与层之间来回横跳改一个字段要动五个文件。根因是先建了文件夹后想数据流。正确的顺序反过来先拿一张纸把一次完整的用户请求从头画到尾。以用户问一个关于内部文档的问题为例数据流是这样的Vue前端发一个POST请求body里带question和session_idFastAPI的route收到后交给service层service层先拿session_id去取历史对话然后把question交给LangChain的retrieverretriever去向量库做相似度检索拿回top-k个文档片段这些片段和问题一起拼成prompt送给大模型大模型返回答案service层把答案、引用的文档片段、本次的token消耗一起打包返回前端拿到后渲染成对话气泡并把引用片段折叠展示。这张图画完文件夹结构自然就出来了api/放routeservices/放业务编排chains/放LangChain相关db/放向量库和会话存储schemas/放Pydantic模型。每个文件夹对应数据流上的一段而不是对应技术类型。这个区别很关键按技术类型分你会得到一堆互不相关的工具函数按数据流分你会得到一条能追下去的链路。2.2 FastAPI的目录不是越细越好新手容易犯的另一个错是把FastAPI项目拆得过于细碎routers/chat.py、routers/chat_utils.py、routers/chat_helpers.py……最后自己都找不到某个函数在哪。我的经验是按业务域拆不按函数类型拆。这个项目里业务域其实就三个对话chat、知识库管理knowledge、系统状态health。所以api/下面三个文件就够了每个文件里把相关的route写在一起超过300行再考虑拆。# api/chat.py from fastapi import APIRouter, Depends from schemas.chat import ChatRequest, ChatResponse from services.chat_service import ChatService router APIRouter(prefix/api/chat, tags[chat]) router.post(, response_modelChatResponse) async def chat(req: ChatRequest, svc: ChatService Depends()): return await svc.handle(req)注意这里用了Depends注入service而不是在route里直接new一个。这样做的好处是service的依赖比如向量库客户端、大模型客户端可以在一个地方统一管理测试的时候也能替换成mock。很多人图省事直接在route里写业务逻辑写到后面route文件上千行改一个检索参数要翻半天。2.3 配置管理别把密钥写死在代码里这一条我必须单独拎出来说。我见过有人把API key直接写在main.py里然后推到公开仓库第二天就收到账单。正确做法是用pydantic-settings读环境变量本地开发用.env文件.env必须进.gitignore。# core/config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): llm_api_key: str llm_base_url: str https://api.example.com/v1 embedding_model: str text-embedding-3-small vector_store_path: str ./data/vectors redis_url: str redis://localhost:6379/0 class Config: env_file .env settings Settings()这里有个细节llm_base_url给了默认值但llm_api_key没有。这是故意的——密钥这种东西必须显式提供缺了就直接启动报错而不是用一个空字符串悄悄跑起来然后在运行时才炸。让错误尽早暴露是省时间的最有效手段。3. RAG这条链检索质量决定一切3.1 文档切分不是随便切RAG最容易翻车的地方不是模型是切分。我一开始图省事按固定500字符切结果切出来的片段经常从句子中间断开检索出来的内容读都读不通。后来改成按语义切分先按段落分段落超过阈值再按句子分句子还超才硬切。# chains/splitter.py from langchain.text_splitter import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size800, chunk_overlap120, separators[\n\n, \n, 。, , , , , ], )chunk_overlap这个参数很多人忽略但它很关键。如果两个相关句子正好被切在边界上没有overlap的话检索时可能两个片段都只命中一半拼起来语义不完整。120字符的overlap大概能覆盖一到两句话实测下来召回率比不重叠高不少。separators里我把中文标点放在英文空格前面因为中文文档里句子边界主要是中文标点这个顺序能让切分更贴合中文习惯。3.2 向量库选型本地优先别急着上集群选向量库的时候很多人一上来就研究Milvus、Weaviate这些分布式方案。我的建议是数据量在十万条以下直接用Chroma或FAISS本地跑。原因很简单本地库零运维、启动快、调试方便而且这个阶段你根本用不到分布式的那点性能。等数据量真的上来了再迁移也不迟LangChain的VectorStore接口是统一的换实现基本只改一行。# db/vector_store.py from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings def get_vector_store(): return Chroma( collection_nameknowledge, embedding_functionOpenAIEmbeddings(modelsettings.embedding_model), persist_directorysettings.vector_store_path, )这里有个坑persist_directory指向的目录如果多个进程同时写Chroma会报锁冲突。开发时用uvicorn --reload热重载经常会出现旧进程还没退、新进程已经起来的情况然后就卡住了。解决办法是热重载时把persist关掉或者用单worker启动。生产环境用多worker的话就得换成支持并发的方案这是后话。3.3 检索之后必须加一层重排只用向量相似度检索效果往往差强人意。因为向量检索本质是语义相近但语义相近不等于能回答问题。我实测过一个案例用户问退款流程要几天向量检索把一段讲退款政策适用范围的文档排在了第一位而真正讲到账时间的文档排在第四。这时候就需要重排rerank把真正相关的往前拉。重排有两种做法一种是用专门的重排模型比如bge-reranker系列把query和每个候选片段拼起来打分另一种是让大模型自己判断相关性。前者快且便宜后者准但贵。我的做法是先用向量检索召回top-20再用重排模型筛出top-5送给大模型。这样既控制了成本又保证了质量。# chains/retriever.py def retrieve_with_rerank(query: str, top_k: int 5): candidates vector_store.similarity_search(query, k20) scored reranker.score(query, [c.page_content for c in candidates]) ranked sorted(zip(candidates, scored), keylambda x: -x[1]) return [c for c, _ in ranked[:top_k]]注意重排模型和Embedding模型最好用同一家的不然打分尺度可能对不上排序结果会不稳定。3.4 引用溯源让答案可验证RAG系统如果不给引用用户没法判断答案是不是编的。所以返回结果里必须带上这段话来自哪个文档的哪一段。实现上在切分的时候给每个chunk打上source和chunk_id的metadata检索回来之后把这些metadata一起返回给前端。chunk.metadata {source: file_name, chunk_id: idx, page: page_num}前端拿到之后在答案下面列出引用来源用户点一下能跳到原文。这个功能看起来小但它是RAG系统可信度的基石。我做过一个小范围测试带引用和不带引用的两个版本用户对带引用版本的信任度明显更高即使答案内容完全一样。4. Agent与Harness从会聊天到能干活4.1 Agent的本质是让模型决定下一步调什么普通对话是用户问模型答一问一答就结束了。Agent不一样它是用户给个目标模型自己决定先做什么、再做什么、什么时候停。比如用户说帮我查一下上个月的销售数据并生成一份简报Agent会先调一个查数据库的工具拿到数据后再调一个生成文档的工具最后把结果返回。LangChain里实现Agent的核心是AgentExecutor加一组Tool。每个Tool就是一个函数带描述模型根据描述决定什么时候调、传什么参数。# chains/agent.py from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain.tools import tool tool def query_sales(month: str) - str: 查询指定月份的销售数据month格式为YYYY-MM return db.query_sales(month) tool def make_report(data: str) - str: 根据销售数据生成简报 return llm.invoke(f根据以下数据生成简报{data}) tools [query_sales, make_report] agent create_openai_tools_agent(llm, tools, prompt) executor AgentExecutor(agentagent, toolstools, max_iterations5)max_iterations这个参数一定要设。我踩过一次坑模型陷入循环反复调同一个工具因为没设上限请求一直不返回最后超时。设成5之后最多五步就强制停返回当前结果。Agent的稳定性很大程度上取决于你有没有给它设边界。4.2 Harness不是测试框架是可观测性基建很多人把Harness理解成单元测试其实它更像是一套让Agent行为可复现、可对比、可回归的基建。具体来说它要解决三个问题第一同一个输入今天跑和明天跑结果不一样怎么定位是模型变了还是代码变了第二改了prompt之后怎么知道是变好了还是变差了第三线上出问题时怎么把当时的完整上下文捞出来复现。我的做法是给每次Agent调用打一个trace记录输入、每一步的工具调用、每步的输出、最终结果、耗时、token消耗。这些trace存到本地文件或数据库出问题时直接按session_id捞出来。# harness/tracer.py import json, time, uuid class Tracer: def __init__(self, path./data/traces): self.path path def log(self, session_id, step, payload): record { ts: time.time(), session_id: session_id, step: step, payload: payload, } with open(f{self.path}/{session_id}.jsonl, a) as f: f.write(json.dumps(record, ensure_asciiFalse) \n)用jsonl格式是因为它追加写方便而且一行一条记录出问题时grep一下就能定位。这个trace机制看起来简单但它是后面做回归测试的基础——有了历史trace你改完prompt之后可以拿同一批输入重跑对比结果差异。4.3 回归测试用固定用例守住底线有了trace就可以建一个回归测试集。我一般会攒20到50个典型问题每个问题标注期望命中的文档和期望答案的关键点。每次改完prompt或换模型跑一遍这个测试集看命中率和关键点覆盖率有没有下降。# harness/regression.py def run_regression(cases): results [] for case in cases: answer agent_executor.invoke({input: case[question]}) hit any(doc in answer[source] for doc in case[expected_docs]) results.append({q: case[question], hit: hit, answer: answer}) hit_rate sum(r[hit] for r in results) / len(results) return hit_rate, results这个测试集不需要很大但必须覆盖你系统的主要使用场景。我见过有人攒了500个用例跑一次要半小时结果没人愿意跑最后形同虚设。20个用例跑一次两分钟改完随手就跑这才是能坚持下去的规模。5. 前后端联调那些文档不会告诉你的细节5.1 流式输出不是可选项大模型生成一段300字的回答非流式的话用户要盯着转圈等五六秒体验很差。流式输出让字一个一个蹦出来感知延迟直接降到几百毫秒。FastAPI这边用StreamingResponse前端用EventSource或fetch的流式读取。# api/chat.py from fastapi.responses import StreamingResponse router.post(/stream) async def chat_stream(req: ChatRequest): async def gen(): async for token in svc.stream_handle(req): yield fdata: {json.dumps({token: token})}\n\n return StreamingResponse(gen(), media_typetext/event-stream)这里有个坑SSE格式要求每条消息以data:开头、以两个换行结尾少一个换行前端就收不到。我第一次写的时候漏了\n\n前端一直没反应查了半小时才发现是格式问题。另外如果中间有代理层记得关掉它的缓冲不然流式会被攒成一坨再发。5.2 跨域和超时开发环境的两个老大难Vue3开发服务器默认跑在5173端口FastAPI跑在8000跨域是必然的。最省事的做法是在vite.config.ts里配proxy把/api转发到8000这样前端代码里直接写相对路径生产环境也不用改。// vite.config.ts export default defineConfig({ server: { proxy: { /api: { target: http://localhost:8000, changeOrigin: true } } } })超时问题更隐蔽。Agent跑一个复杂任务可能要30秒以上而很多HTTP客户端默认超时是30秒正好卡在边界上。我的做法是前端fetch显式设一个更长的超时后端也把uvicorn的--timeout-keep-alive调大。另外长任务最好改成提交任务返回task_id前端轮询结果的模式避免连接一直挂着。5.3 会话状态存哪内存、Redis还是数据库开发阶段用内存字典存会话最简单但一重启就没了而且多worker之间不共享。上生产就得用Redis。我的建议是从第一天就用Redis哪怕本地开发也起一个Redis容器因为会话存储的接口一旦定下来后面换实现很麻烦。# db/session.py import redis, json r redis.from_url(settings.redis_url) def get_history(session_id: str, limit: int 10): raw r.lrange(fchat:{session_id}, -limit, -1) return [json.loads(x) for x in raw] def append_message(session_id: str, role: str, content: str): r.rpush(fchat:{session_id}, json.dumps({role: role, content: content})) r.expire(fchat:{session_id}, 86400)expire设一天避免会话数据无限堆积。历史对话只取最近10条因为全量塞进prompt既贵又容易超上下文窗口。这里有个取舍取太少模型记不住上下文取太多成本和延迟都上去了。10条是我实测下来比较平衡的值你可以根据自己的场景调。6. 部署与性能从能跑到跑得稳6.1 容器化一个Dockerfile搞定后端后端打包成镜像关键是分层构建把依赖安装和代码拷贝分开这样改代码时不用重装依赖构建快很多。FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]注意--no-cache-dir能显著减小镜像体积。另外向量库的持久化目录要挂volume不然容器一重建数据就没了。前端用npm run build出静态文件交给Nginx托管Nginx同时反代/api到后端容器。6.2 缓存省钱的直接手段大模型调用是花钱的能缓存就缓存。两个层面可以缓存一是Embedding结果同一段文本不要重复算二是问答结果相同问题在短时间内直接返回缓存。# services/cache.py import hashlib def cache_key(text: str) - str: return emb: hashlib.md5(text.encode()).hexdigest() def get_or_compute_embedding(text: str): key cache_key(text) cached r.get(key) if cached: return json.loads(cached) vec embedding_model.embed_query(text) r.set(key, json.dumps(vec), ex86400) return vecEmbedding缓存命中率通常很高因为知识库文档是固定的重复入库时能省掉大量计算。问答缓存要小心因为同一个问题在不同上下文下答案可能不同所以缓存key里最好带上会话的关键状态或者只对无状态的单轮问答做缓存。6.3 监控至少要知道哪里慢上线之后你至少要知道三件事每个接口的P95延迟、大模型调用的失败率、向量检索的耗时。这三个指标能覆盖大部分问题。实现上用一个中间件记录每个请求的耗时打到日志里再用简单的脚本聚合。# middleware/timing.py app.middleware(http) async def timing(request, call_next): start time.time() response await call_next(request) cost time.time() - start logger.info(f{request.url.path} cost{cost:.3f}s) return response日志里带上路径和耗时出问题时grep一下就能看出是哪个接口慢。如果发现大模型调用占了总耗时的80%那优化方向就很明确了——要么换更快的模型要么加缓存要么把非关键步骤异步化。7. 我踩过的几个印象最深的坑第一个坑是上下文窗口超限。有一次用户传了一篇很长的文档进来我直接把全文塞进prompt结果接口报错说超过最大token。后来改成先切分、只取相关片段问题解决。教训是任何进prompt的内容都要先估算token不能想当然。第二个坑是异步阻塞。我在async的route里调了一个同步的向量库查询结果整个事件循环被卡住并发一上来所有请求都排队。解决办法是用run_in_executor把同步调用丢到线程池或者直接用支持异步的客户端。第三个坑是重排模型和Embedding模型维度不匹配。我一开始混用了两家的模型检索结果排序完全乱套排查了很久才发现是打分尺度不一致。后来统一用同一家的模型问题消失。第四个坑是前端流式渲染的节流。字一个一个蹦出来如果每个token都触发一次Vue的响应式更新页面会卡。解决办法是攒几个token再更新一次DOM或者用requestAnimationFrame批量刷新。这些坑的共同点是文档里不会写只有真跑起来才会遇到。所以我一直建议学AI应用开发最好的方式不是看教程而是自己搭一个能跑的小系统然后不断往里加需求坑自然就踩出来了。8. 后续可以怎么扩展这套骨架搭起来之后扩展方向其实很多。想做多模态就在检索层加上图片的向量化前端加图片上传和展示想做多租户就在向量库和会话存储里加tenant_id字段检索时按租户过滤想做工作流编排就把Agent的线性执行改成DAG支持并行和条件分支。我个人最推荐的扩展方向是把Harness做厚。因为随着系统变复杂模型、prompt、工具、检索策略都在变没有一套可靠的回归和观测机制你根本不知道某次改动是让系统变好了还是变差了。我现在的习惯是任何影响输出的改动都必须先跑一遍回归测试集看指标有没有下降再决定要不要合入。这个习惯帮我避免了好几次改了一个地方另一个地方悄悄坏掉的事故。最后分享一个小技巧把每次线上出问题的完整trace存下来攒成一个事故案例库。下次遇到类似问题先翻这个库往往能直接找到答案。这个库比任何文档都值钱因为它是你自己系统真实踩过的坑。