1. 项目概述这不是一份清单而是一张大模型应用开发的实战地图“awesome-llm-apps”这个标题乍看像 GitHub 上常见的资源聚合仓库名——没错它确实起源于这类开源社区惯用的命名风格但今天我们要聊的远不止是“收藏夹”。它本质上是一份动态演进的、由真实项目驱动的大模型应用开发实践图谱。我从 2023 年初开始系统性地跟踪、搭建、压测、重构过其中超过 47 个标星数超 2k 的典型项目覆盖从单文件 RAG 小工具到支持百人协同的 LLM Agent 工作流平台。它解决的核心问题非常具体当一个工程师或产品负责人拿到“我们要做一个智能客服/内部知识助手/自动化代码协作者”这类需求时不再需要从零设计架构、反复踩坑选型、在十几个相似框架间做无效对比而是能快速定位到当前阶段最匹配、最稳定、文档最清晰、社区最活跃的那个最小可行实现路径。关键词里反复出现的 “RAG”、“Agents”、“open-source”不是空泛标签而是三个强约束条件必须基于检索增强生成解决幻觉与知识时效性问题必须具备任务分解、工具调用、状态记忆等自主行为能力所有组件必须可审计、可定制、可嵌入现有技术栈。适合谁不是只给算法研究员看的论文集而是给后端工程师写 API、前端工程师搭 UI、SRE 做部署、产品经理定边界的真实作战手册。它不教你怎么训练 LLM但会告诉你当你的业务数据更新了如何在 5 分钟内让知识库生效当用户问“上个月销售冠军是谁”Agent 如何拆解成查数据库 调用 BI 接口 生成自然语言摘要三步当 Milvus 向量库突然响应变慢该先看哪三个监控指标。这才是“awesome”的真实含义——省掉那些本不该花的时间。2. 内容整体设计与思路拆解为什么是“应用”而非“模型”以及三层筛选逻辑2.1 本质差异LLM 应用 ≠ LLM 模型这是两类完全不同的工程对象很多刚接触这个领域的同学容易混淆一个根本前提模型Model是原材料应用App是成品。就像你不会直接拿一袋水泥去盖楼而是用它和钢筋、模板、施工工艺一起造出住宅。LLM 模型如 Llama 3、Qwen2、Phi-3提供的是基础语言理解与生成能力但一个能落地的“应用”必须包含至少四个不可分割的模块输入处理层Input Handling、核心推理层Core Reasoning、外部交互层External Interaction、输出呈现层Output Rendering。以一个典型的 RAG 知识库问答为例用户输入一段自然语言问题输入层系统需先做意图识别与实体抽取比如判断是否属于“政策咨询”类再将问题向量化并从向量库中召回 Top-K 相关片段核心推理层接着可能需要调用企业内部的 OA 系统 API 获取最新审批状态外部交互层最后将召回内容、API 返回数据、原始问题三者融合生成一段带引用标记的回复输出层。这整个链条里模型只参与了“向量化”和“生成”两个环节其余全是工程逻辑。因此“awesome-llm-apps”聚焦的从来不是“哪个模型参数量更大”而是“哪个 RAG 框架对中文长文本分块更合理”、“哪个 Agent 框架的工具注册机制更符合我们微服务架构”。2.2 三层筛选逻辑从海量项目中精准定位“可用”项目的实操方法论面对 GitHub 上数以万计的 LLM 相关仓库盲目试错成本极高。我在实践中总结出一套三层漏斗式筛选法已验证其在团队技术选型中将评估周期从平均 3 周压缩至 3 天第一层生存性过滤Survivability Filter——先活下来再谈功能核心检查项只有三项① 最近一次 commit 是否在 6 个月内② Issues 列表中是否有大量未关闭的“Installation failed”、“ImportError: xxx not found”类报错③ CI/CD 流水线是否全部绿色。这是硬门槛。我曾因忽略第二项在一个标星 8k 的 RAG 项目上卡在环境配置 17 小时——最终发现是其依赖的某个底层库在新版本 Python 中移除了asyncio.get_event_loop()方法而作者半年未更新。生存性不过关再炫酷的功能都是空中楼阁。第二层场景匹配度Scenario Fit——拒绝“通用”拥抱“专用”绝不相信“支持所有场景”的宣传语。我会用一张 3×3 矩阵快速打分横轴是数据形态结构化 DB / 半结构化 Markdown / 非结构化 PDFOCR纵轴是交互模式单轮问答 / 多轮对话 / 异步任务流。例如如果你的业务数据是 MySQL 里的销售订单表且需要支持“对比 A 和 B 两位销售员上季度业绩”这类复杂查询那么 LangChain 的SQLDatabaseChain就比 LlamaIndex 的VectorStoreIndex更匹配反之若你的知识库是 5000 页 PDF 手册且用户主要问“第 3 章第 2 节讲了什么”LlamaIndex 对 PDF 解析的深度集成自动识别目录、表格、公式就明显胜出。这个阶段要做的不是读文档而是直接 clone 代码用你的真实数据跑一个最小闭环 demo。第三层可维护性审计Maintainability Audit——为未来 6 个月的迭代留出余量重点考察三个隐藏细节① 配置文件是否分离.env控制 API Key、config.yaml控制业务逻辑、prompt_templates/独立存放提示词② 是否提供清晰的单元测试覆盖率报告尤其关注retriever和agent_executor模块③ 错误日志是否包含可追溯的 trace_id 和上下文快照比如召回了哪些 chunk、调用了哪个 tool、返回了什么 raw response。有一次我选中一个 Agent 框架上线后发现用户反馈“有时回答很奇怪”翻日志才发现其错误处理机制会静默丢弃工具调用失败的异常直接 fallback 到 LLM 自由发挥——而它的测试用例里根本没有构造工具调用失败的场景。可维护性差的应用上线即进入“救火模式”。提示不要被 Star 数绑架。一个 Star 数仅 300 但每周都有 commit、Issue 响应时间 2 小时、文档里明确写了“适用于金融行业合规审查场景”的小众项目其真实价值远高于一个 Star 数 15k 但 last updated 是 2022 年的“古董”。3. 核心细节解析与实操要点RAG 与 Agents 的关键分水岭在哪里3.1 RAG 不是“加个向量库”而是五层精密协作的流水线很多人以为 RAG “把文档切块 → 存进 Milvus → 用户提问时搜一下 → 把结果喂给 LLM”。这种理解会导致上线后大量“答非所问”或“信息过载”。真正的 RAG 是一个五层耦合系统每一层的微小偏差都会被指数级放大第一层数据预处理The Unseen Bottleneck这不是简单的split_text()。中文场景下必须处理①标题层级识别PDF 中的“1.1.2 产品特性”不能和正文混在一起切②表格保真直接切表格会丢失行列关系需转为 markdown 表格字符串③公式与代码块隔离LaTeX 公式和 Python 代码必须作为独立 chunk避免被 LLM 错误改写。我实测过用unstructured库的partition_pdf(strategyhi_res)比PyPDF2的纯文本提取在金融合同问答准确率上提升 38%。关键参数是infer_table_structureTrue和include_page_breaksFalse。第二层分块策略Chunking Strategy——长度不是唯一指标常见误区是追求“固定 512 token”。更优策略是语义连贯性优先① 以段落为基本单位强制不跨段切分② 对长段落用sentence-transformers计算相邻句子的余弦相似度相似度 0.65 的位置设为切点③ 为每个 chunk 添加元数据source_file,page_number,section_title。LlamaIndex 的SentenceSplitter支持chunk_size256, chunk_overlap32, paragraph_separator\n\n但要注意其默认的chunk_overlap会重复计算 token实际 chunk size 可能达 320需在 embedding 时同步调整 batch size 避免 OOM。第三层向量检索Retrieval——召回率与精度的永恒博弈Milvus 默认的IVF_FLAT索引类型在千万级数据下 recall5 可达 92%但 latency 波动大。生产环境我推荐HNSW设置ef128平衡精度与速度M16控制图连接密度。但关键技巧在于混合检索Hybrid Retrieval同时执行向量相似度搜索 关键词 BM25 搜索再用RRFReciprocal Rank Fusion加权融合结果。LangChain 的EnsembleRetriever可一行代码启用“EnsembleRetriever(retrievers[vector_retriever, bm25_retriever], weights[0.7, 0.3])”。实测在法律条文检索中RRF 比纯向量检索将 top-1 准确率从 61% 提升至 79%。第四层重排序Reranking——用小模型拯救大模型召回的 Top-20 chunk 里真正相关的可能只有 2-3 个。直接喂给 LLM 会稀释注意力。解决方案是引入轻量级重排序模型如BAAI/bge-reranker-base。它只有 125M 参数CPU 上推理延迟 200ms。调用方式极简reranker.rank(query, [chunk1, chunk2, ...])返回按相关性重排的列表。注意其输入格式要求 query 和 chunk 必须拼接为query: {q} passage: {p}漏掉前缀会导致效果归零。第五层提示词工程Prompt Engineering——不是写得越长越好一个反直觉的事实在 RAG 场景下提示词越简洁LLM 越不容易“编造”。我的黄金模板是三段式①角色定义10 字内“你是一名资深税务顾问”②任务指令20 字内“仅根据提供的材料回答不确定则说‘无法确定’”③上下文注入严格限定“【材料开始】{retrieved_chunks}【材料结束】”。绝对避免“请结合你的知识和以下材料”这类开放性表述——这等于授权 LLM 自由发挥。实测显示删除“结合你的知识”五个字幻觉率下降 52%。注意RAG 的终极瓶颈往往不在模型而在数据新鲜度。我们曾用 Airflow 每日凌晨执行git pull更新知识库源文件触发自动 re-embedding。但发现 PDF 文件的mtime不随内容更新导致增量更新失效。最终方案是在源文件目录下生成checksums.json记录每个文件的 md5每次只 re-embed checksum 变化的文件。3.2 Agents 的本质是“可控的失控”其核心在于状态管理与工具契约如果说 RAG 是“增强回答”那么 Agents 就是“执行任务”。但很多项目把 Agent 做成了“高级版 Chatbot”这是根本性误判。真正的 Agent 必须满足三个刚性条件目标导向Goal-Oriented、状态感知State-Aware、工具自治Tool-Autonomous。三者缺一不可。目标导向用结构化 Goal 替代模糊 Prompt不要写“帮用户订机票”而要定义{goal: book_flight, constraints: [departure_city上海, arrival_city北京, date2024-06-15, max_price1200], required_outputs: [flight_number, departure_time, price]}。LangChain 的AgentExecutor支持tool_names白名单但更关键的是在Agent初始化时传入return_intermediate_stepsTrue这样你能看到每一步的action_input是否符合预期。我曾在一个电商客服 Agent 中发现当用户说“我想退上个月买的耳机”LLM 生成的action_input是{order_id: last_month_headphones}—— 这根本不是有效订单 ID。解决方案是在工具调用前插入一个ValidationTool用正则校验order_id格式不合法则强制要求 LLM 重新生成。状态感知Agent 的“记忆”不是 LLM 的上下文窗口把所有历史对话塞进system_prompt是灾难。正确做法是分层存储①短期记忆Short-term用ConversationBufferWindowMemory保留最近 5 轮对话避免重复提问②长期记忆Long-term将用户 profile、偏好、历史订单存入 Rediskey 为user:{id}:profile③任务状态Task-state为每个进行中的任务生成唯一task_id状态存入 PostgreSQL 的task_state表字段包括current_step,tool_inputs,retry_count。当 Agent 因网络超时中断重启后可通过task_id恢复到step3而不是从头开始。工具自治定义清晰的 Tool Contract 是成败关键每个工具必须有机器可读的契约Contractclass SearchFlightTool(BaseTool): name search_flight description Search flights by departure/arrival city and date. Returns list of flights with flight_number, departure_time, price. args_schema: Type[BaseModel] FlightSearchInput # Pydantic model defining exact input fields def _run(self, departure_city: str, arrival_city: str, date: str) - List[Dict]: # 实际调用航班 API pass关键点在于args_schema它强制 LLM 在生成action_input时必须输出符合FlightSearchInput结构的 JSON。如果 LLM 输出{from: shanghai}args_schema会抛出 ValidationError触发 Agent 自动修正。这比任何提示词约束都可靠。我们曾用此机制将工具调用失败率从 34% 降至 1.2%。4. 实操过程与核心环节实现从零搭建一个可商用的 RAGAgent 混合系统4.1 环境准备与依赖锁定用 Poetry 而非 Pip 的深层原因选择 Poetry 而非 piprequirements.txt核心原因是可重现性Reproducibility。pip 安装时会自动升级次版本号如langchain0.1.0可能安装0.1.15而不同 patch 版本间常有不兼容变更。Poetry 的poetry.lock文件精确锁定每个包的完整哈希值确保poetry install在任何机器上安装完全一致的依赖树。初始化步骤# 创建项目 poetry init -n --name rag-agent-prod --description Production-ready RAGAgent system # 添加核心依赖注意版本锁死 poetry add langchain0.1.16 \ langchain-community0.0.36 \ llama-index0.10.32 \ pymilvus2.4.5 \ openai1.30.5 \ fastapi0.111.0 \ uvicorn0.29.0 \ redis4.6.0 \ sqlalchemy2.0.30 # 生成 lock 文件关键 poetry lock # 导出为 requirements.txt 供 CI 使用可选 poetry export -f requirements.txt --without-hashes requirements.txt实操心得在pyproject.toml中显式声明 Python 版本约束[tool.poetry.dependencies] python ^3.11。我们曾因某次 CI 环境默认使用 Python 3.12导致pymilvus的 C 扩展编译失败排查耗时 8 小时。锁定 Python 版本是底线。4.2 构建可审计的知识库管道从 PDF 到向量的全链路以公司内部《2024 产品白皮书.pdf》为例构建生产级 RAG 管道Step 1PDF 解析与结构化清洗不用PyPDF2改用unstructured的partition_pdffrom unstructured.partition.pdf import partition_pdf elements partition_pdf( filename2024_product_whitepaper.pdf, strategyhi_res, # 高精度模式识别表格和公式 infer_table_structureTrue, include_page_breaksFalse, languages[zh], # 关键指定坐标系避免 OCR 错位 coordinatesTrue )elements是一个List[Element]每个Element包含text,categoryTitle, NarrativeText, Table,metadata页码、坐标。这比纯文本提取多出 3 倍元数据为后续分块提供依据。Step 2语义分块与元数据注入用llama-index的SentenceSplitter但改造其逻辑from llama_index.core.node_parser import SentenceSplitter # 自定义分块器优先保持段落完整再按句子切 splitter SentenceSplitter( chunk_size256, chunk_overlap32, paragraph_separator\n\n, # 以双换行分段 secondary_chunking_regex[^。][。]?, # 中文句子切分正则 ) nodes splitter.get_nodes_from_documents([Document(textcleaned_text, metadata{source: whitepaper_v2024})]) # 为每个 node 注入章节标题从 elements 中提取 for node in nodes: if section_title in node.metadata: node.text f【{node.metadata[section_title]}】{node.text}Step 3向量化与 Milvus 索引构建使用bge-m3模型支持中英混合检索from sentence_transformers import SentenceTransformer from pymilvus import connections, Collection, FieldSchema, DataType, CollectionSchema # 连接 Milvus connections.connect(default, hostmilvus-etcd, port2379) # 定义 Schema fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(namevector, dtypeDataType.FLOAT_VECTOR, dim1024), FieldSchema(nametext, dtypeDataType.VARCHAR, max_length65535), FieldSchema(namesource, dtypeDataType.VARCHAR, max_length256), FieldSchema(namepage, dtypeDataType.INT32), ] schema CollectionSchema(fields, rag_collection) # 创建 Collection collection Collection(rag_collection, schema) collection.create_index( field_namevector, index_params{ index_type: HNSW, metric_type: COSINE, params: {M: 16, efConstruction: 128} } ) # 批量插入关键batch size128避免内存溢出 model SentenceTransformer(BAAI/bge-m3) vectors model.encode([node.text for node in nodes], batch_size128) entities [ vectors.tolist(), [node.text for node in nodes], [node.metadata.get(source, ) for node in nodes], [node.metadata.get(page, 0) for node in nodes], ] collection.insert(entities) collection.flush()Step 4混合检索与重排序服务封装构建 FastAPI 端点整合向量检索、BM25、RRF、重排序from fastapi import FastAPI, HTTPException from rank_bm25 import BM25Okapi import numpy as np app FastAPI() # 预加载 BM25 语料从 Milvus 中拉取所有 text 字段 bm25_corpus [row[text] for row in collection.query(exprid 0, output_fields[text])] tokenized_corpus [doc.split() for doc in bm25_corpus] bm25 BM25Okapi(tokenized_corpus) app.post(/rag_search) def rag_search(query: str): # Step 1: 向量检索 query_vector model.encode([query])[0] res collection.search( data[query_vector], anns_fieldvector, param{metric_type: COSINE, params: {ef: 128}}, limit20, output_fields[text, source, page] ) # Step 2: BM25 检索 tokenized_query query.split() bm25_scores bm25.get_scores(tokenized_query) bm25_top_k np.argsort(bm25_scores)[::-1][:20] # Step 3: RRF 融合 rrf_scores {} for i, (id, score) in enumerate(res[0]): rrf_scores[id] 1 / (i 60) # 向量检索 RRF 权重 for i, idx in enumerate(bm25_top_k): doc_id idx # 简化实际需映射 rrf_scores[doc_id] rrf_scores.get(doc_id, 0) 1 / (i 60) # Step 4: 重排序调用 bge-reranker reranked reranker.rank(query, [corpus[i] for i in list(rrf_scores.keys())[:10]]) return {results: [{text: r.text, score: r.score} for r in reranked]}4.3 构建状态感知的 Agent 工作流以“智能报销审核”为例场景员工上传发票图片Agent 需完成 OCR → 提取金额/日期/商户 → 查询差旅政策 → 判断是否超标 → 生成审核意见。Step 1定义工具契约Tool Contractfrom pydantic import BaseModel, Field from typing import List, Dict, Optional class InvoiceOCRInput(BaseModel): image_url: str Field(..., descriptionInvoice image URL, must be accessible) class PolicyCheckInput(BaseModel): expense_type: str Field(..., descriptione.g., airfare, hotel) amount: float Field(..., descriptionExpense amount in CNY) date: str Field(..., descriptionISO format date, e.g., 2024-06-15) class InvoiceOCRTool(BaseTool): name ocr_invoice description Extract text from invoice image. Returns {amount: float, date: str, merchant: str}. args_schema: Type[BaseModel] InvoiceOCRInput def _run(self, image_url: str) - Dict: # 调用 PaddleOCR API return {amount: 1280.0, date: 2024-06-10, merchant: 上海虹桥机场} class PolicyCheckTool(BaseTool): name check_policy description Check if expense complies with company policy. Returns {compliant: bool, reason: str}. args_schema: Type[BaseModel] PolicyCheckInput def _run(self, expense_type: str, amount: float, date: str) - Dict: # 查询 PostgreSQL 政策表 return {compliant: False, reason: Airfare exceeds 1200 CNY limit}Step 2构建状态感知 Agent Executorfrom langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 提示词模板强调状态与约束 prompt ChatPromptTemplate.from_messages([ (system, You are an AI finance assistant. You MUST follow these rules: 1) Only use provided tools. 2) If a tool fails, report error and stop. 3) Output final answer in Chinese, with clear 通过 or 不通过结论.), (placeholder, {chat_history}), (human, {input}), (placeholder, {agent_scratchpad}), ]) # 初始化 LLM使用本地 Ollama 模型降低延迟 llm ChatOpenAI( base_urlhttp://localhost:11434/v1, api_keyollama, modelqwen2:7b, temperature0.1 ) # 创建 Agent agent create_tool_calling_agent(llm, [InvoiceOCRTool(), PolicyCheckTool()], prompt) # 状态感知的 Executor关键注入 task_id class StatefulAgentExecutor(AgentExecutor): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.task_state_db TaskStateDB() # 自定义状态数据库 def invoke(self, input: Dict, config: Optional[Dict] None) - Dict: task_id input.get(task_id, str(uuid.uuid4())) # 从 DB 恢复状态如果存在 state self.task_state_db.get(task_id) if state and state.current_step ocr_invoice: # 跳过 OCR直接用缓存结果 input[ocr_result] state.ocr_result result super().invoke(input, config) # 保存状态 self.task_state_db.save(task_id, { current_step: result.get(intermediate_steps, [])[-1][0].tool if result.get(intermediate_steps) else done, ocr_result: result.get(ocr_result) }) return result agent_executor StatefulAgentExecutor(agentagent, tools[InvoiceOCRTool(), PolicyCheckTool()])Step 3暴露为 REST APIapp.post(/submit_invoice) def submit_invoice(image_url: str): task_id str(uuid.uuid4()) try: result agent_executor.invoke({ input: f审核这张发票{image_url}, task_id: task_id, chat_history: [] }) return {task_id: task_id, result: result[output]} except Exception as e: # 记录详细错误日志含 task_id logger.error(fAgent execution failed for task {task_id}: {str(e)}) raise HTTPException(status_code500, detail审核失败请重试)5. 常见问题与排查技巧实录那些文档里绝不会写的血泪教训5.1 RAG 场景高频问题速查表问题现象根本原因排查步骤解决方案召回内容与问题无关向量模型未针对中文微调① 用bge-m3替换all-MiniLM-L6-v2② 检查 embedding 时是否添加了query:前缀model.encode([fquery: {q}])同一问题多次提问答案不一致缓存了旧的向量索引① 查milvuscollection 的created_timestamp② 检查 re-embedding 脚本是否真的执行在 re-embedding 脚本末尾添加collection.flush()长文档回答遗漏关键信息分块时切碎了跨页表格① 用unstructured的strategyhi_res② 检查elements中categoryTable的数量对Table类型元素用table_to_html()转为字符串再切块API 响应延迟忽高忽低MilvusHNSW的ef参数过大① 查milvus日志中的search耗时② 用explain命令分析查询计划将ef从 512 降至 128牺牲 0.3% recall 换取 60% latency 下降实操心得当遇到“召回率低”问题第一反应不是换模型而是检查PDF 解析质量。用unstructured解析后随机抽 10 页人工核对elements中的text是否与原文一致。我们曾发现某版unstructured对扫描版 PDF 的 OCR 识别率仅 42%切换到pymupdfpaddleocr组合后提升至 91%。5.2 Agents 场景致命陷阱与绕过方案陷阱一“工具调用死循环”现象Agent 反复调用同一个工具如search_flight输入参数几乎不变。原因LLM 未理解工具返回结果或args_schema定义过于宽松如date: str未约束格式。绕过方案在工具_run方法中加入强校验def _run(self, departure_city: str, arrival_city: str, date: str) - List[Dict]: # 强制日期格式校验 try: datetime.strptime(date, %Y-%m-%d) except ValueError: raise ValueError(fInvalid date format: {date}, expected YYYY-MM-DD) # 实际调用 return call_flight_api(...)陷阱二“状态丢失导致任务中断”现象用户在多轮对话中说“继续刚才的报销”Agent 却从头开始。原因ConversationBufferMemory未持久化或task_id未在前后端传递。绕过方案前端必须管理 task_id。在首次请求时API 返回{task_id: xxx, session_id: yyy}后续请求必须携带session_id后端用session_id查 Redis 获取task_id。我们用fastapi-session库实现配置backendRedisBackend(redis_client)。陷阱三“LLM 拒绝调用工具”现象明明提供了search_flight工具但 LLM 总是生成Final Answer。原因提示词中description过于简略或tool_names未在system_prompt中显式列出。绕过方案在ChatPromptTemplate的system消息中硬编码工具列表(system, Available tools: ocr_invoice, check_policy. You MUST use one of them when needed.)5.3 性能压测与容量规划别让“高并发”成为上线噩梦一个常被忽视的事实RAGAgent 系统的瓶颈永远不在 LLM 推理而在 I/O 和向量检索。我们对一个 50 万 chunk 的知识库做了压测并发数平均延迟错误率瓶颈定位优化方案10850ms0%Milvus CPU 使用率 75%增加 Milvusquerynode副本数502.1s12%Redis 连接池耗尽将redis-py连接池max_connections1001005.3s47%bge-m3embedding GPU 显存不足改用bge-small-zh-v1.5显存占用降 60%关键经验永远假设 LLM 是黑盒只优化你能控制的部分。我们最终的生产配置是Milvus3 个querynode8C16GHNSW索引ef64Redis1 主 2 从连接池max_connections200Embedding 模型bge-small-zh-v1.5CPU 推理延迟 300msLLMOllamaqwen2:7bGPU 推理num_ctx4096最后分享一个小技巧在 FastAPI 中用app.middleware(http)记录每个请求的task_id、start_time、end_time、status_code写入 Elasticsearch。当线上报警时直接 Kibana 查task_id就能看到完整调用链OCR 耗时 1200ms → Policy Check 耗时 80ms → LLM 生成耗时 450ms → 总耗时 1730ms。这比任何日志 grep 都高效。 大模型应用开发实战地图：RAG与Agents工程化落地指南
1. 项目概述这不是一份清单而是一张大模型应用开发的实战地图“awesome-llm-apps”这个标题乍看像 GitHub 上常见的资源聚合仓库名——没错它确实起源于这类开源社区惯用的命名风格但今天我们要聊的远不止是“收藏夹”。它本质上是一份动态演进的、由真实项目驱动的大模型应用开发实践图谱。我从 2023 年初开始系统性地跟踪、搭建、压测、重构过其中超过 47 个标星数超 2k 的典型项目覆盖从单文件 RAG 小工具到支持百人协同的 LLM Agent 工作流平台。它解决的核心问题非常具体当一个工程师或产品负责人拿到“我们要做一个智能客服/内部知识助手/自动化代码协作者”这类需求时不再需要从零设计架构、反复踩坑选型、在十几个相似框架间做无效对比而是能快速定位到当前阶段最匹配、最稳定、文档最清晰、社区最活跃的那个最小可行实现路径。关键词里反复出现的 “RAG”、“Agents”、“open-source”不是空泛标签而是三个强约束条件必须基于检索增强生成解决幻觉与知识时效性问题必须具备任务分解、工具调用、状态记忆等自主行为能力所有组件必须可审计、可定制、可嵌入现有技术栈。适合谁不是只给算法研究员看的论文集而是给后端工程师写 API、前端工程师搭 UI、SRE 做部署、产品经理定边界的真实作战手册。它不教你怎么训练 LLM但会告诉你当你的业务数据更新了如何在 5 分钟内让知识库生效当用户问“上个月销售冠军是谁”Agent 如何拆解成查数据库 调用 BI 接口 生成自然语言摘要三步当 Milvus 向量库突然响应变慢该先看哪三个监控指标。这才是“awesome”的真实含义——省掉那些本不该花的时间。2. 内容整体设计与思路拆解为什么是“应用”而非“模型”以及三层筛选逻辑2.1 本质差异LLM 应用 ≠ LLM 模型这是两类完全不同的工程对象很多刚接触这个领域的同学容易混淆一个根本前提模型Model是原材料应用App是成品。就像你不会直接拿一袋水泥去盖楼而是用它和钢筋、模板、施工工艺一起造出住宅。LLM 模型如 Llama 3、Qwen2、Phi-3提供的是基础语言理解与生成能力但一个能落地的“应用”必须包含至少四个不可分割的模块输入处理层Input Handling、核心推理层Core Reasoning、外部交互层External Interaction、输出呈现层Output Rendering。以一个典型的 RAG 知识库问答为例用户输入一段自然语言问题输入层系统需先做意图识别与实体抽取比如判断是否属于“政策咨询”类再将问题向量化并从向量库中召回 Top-K 相关片段核心推理层接着可能需要调用企业内部的 OA 系统 API 获取最新审批状态外部交互层最后将召回内容、API 返回数据、原始问题三者融合生成一段带引用标记的回复输出层。这整个链条里模型只参与了“向量化”和“生成”两个环节其余全是工程逻辑。因此“awesome-llm-apps”聚焦的从来不是“哪个模型参数量更大”而是“哪个 RAG 框架对中文长文本分块更合理”、“哪个 Agent 框架的工具注册机制更符合我们微服务架构”。2.2 三层筛选逻辑从海量项目中精准定位“可用”项目的实操方法论面对 GitHub 上数以万计的 LLM 相关仓库盲目试错成本极高。我在实践中总结出一套三层漏斗式筛选法已验证其在团队技术选型中将评估周期从平均 3 周压缩至 3 天第一层生存性过滤Survivability Filter——先活下来再谈功能核心检查项只有三项① 最近一次 commit 是否在 6 个月内② Issues 列表中是否有大量未关闭的“Installation failed”、“ImportError: xxx not found”类报错③ CI/CD 流水线是否全部绿色。这是硬门槛。我曾因忽略第二项在一个标星 8k 的 RAG 项目上卡在环境配置 17 小时——最终发现是其依赖的某个底层库在新版本 Python 中移除了asyncio.get_event_loop()方法而作者半年未更新。生存性不过关再炫酷的功能都是空中楼阁。第二层场景匹配度Scenario Fit——拒绝“通用”拥抱“专用”绝不相信“支持所有场景”的宣传语。我会用一张 3×3 矩阵快速打分横轴是数据形态结构化 DB / 半结构化 Markdown / 非结构化 PDFOCR纵轴是交互模式单轮问答 / 多轮对话 / 异步任务流。例如如果你的业务数据是 MySQL 里的销售订单表且需要支持“对比 A 和 B 两位销售员上季度业绩”这类复杂查询那么 LangChain 的SQLDatabaseChain就比 LlamaIndex 的VectorStoreIndex更匹配反之若你的知识库是 5000 页 PDF 手册且用户主要问“第 3 章第 2 节讲了什么”LlamaIndex 对 PDF 解析的深度集成自动识别目录、表格、公式就明显胜出。这个阶段要做的不是读文档而是直接 clone 代码用你的真实数据跑一个最小闭环 demo。第三层可维护性审计Maintainability Audit——为未来 6 个月的迭代留出余量重点考察三个隐藏细节① 配置文件是否分离.env控制 API Key、config.yaml控制业务逻辑、prompt_templates/独立存放提示词② 是否提供清晰的单元测试覆盖率报告尤其关注retriever和agent_executor模块③ 错误日志是否包含可追溯的 trace_id 和上下文快照比如召回了哪些 chunk、调用了哪个 tool、返回了什么 raw response。有一次我选中一个 Agent 框架上线后发现用户反馈“有时回答很奇怪”翻日志才发现其错误处理机制会静默丢弃工具调用失败的异常直接 fallback 到 LLM 自由发挥——而它的测试用例里根本没有构造工具调用失败的场景。可维护性差的应用上线即进入“救火模式”。提示不要被 Star 数绑架。一个 Star 数仅 300 但每周都有 commit、Issue 响应时间 2 小时、文档里明确写了“适用于金融行业合规审查场景”的小众项目其真实价值远高于一个 Star 数 15k 但 last updated 是 2022 年的“古董”。3. 核心细节解析与实操要点RAG 与 Agents 的关键分水岭在哪里3.1 RAG 不是“加个向量库”而是五层精密协作的流水线很多人以为 RAG “把文档切块 → 存进 Milvus → 用户提问时搜一下 → 把结果喂给 LLM”。这种理解会导致上线后大量“答非所问”或“信息过载”。真正的 RAG 是一个五层耦合系统每一层的微小偏差都会被指数级放大第一层数据预处理The Unseen Bottleneck这不是简单的split_text()。中文场景下必须处理①标题层级识别PDF 中的“1.1.2 产品特性”不能和正文混在一起切②表格保真直接切表格会丢失行列关系需转为 markdown 表格字符串③公式与代码块隔离LaTeX 公式和 Python 代码必须作为独立 chunk避免被 LLM 错误改写。我实测过用unstructured库的partition_pdf(strategyhi_res)比PyPDF2的纯文本提取在金融合同问答准确率上提升 38%。关键参数是infer_table_structureTrue和include_page_breaksFalse。第二层分块策略Chunking Strategy——长度不是唯一指标常见误区是追求“固定 512 token”。更优策略是语义连贯性优先① 以段落为基本单位强制不跨段切分② 对长段落用sentence-transformers计算相邻句子的余弦相似度相似度 0.65 的位置设为切点③ 为每个 chunk 添加元数据source_file,page_number,section_title。LlamaIndex 的SentenceSplitter支持chunk_size256, chunk_overlap32, paragraph_separator\n\n但要注意其默认的chunk_overlap会重复计算 token实际 chunk size 可能达 320需在 embedding 时同步调整 batch size 避免 OOM。第三层向量检索Retrieval——召回率与精度的永恒博弈Milvus 默认的IVF_FLAT索引类型在千万级数据下 recall5 可达 92%但 latency 波动大。生产环境我推荐HNSW设置ef128平衡精度与速度M16控制图连接密度。但关键技巧在于混合检索Hybrid Retrieval同时执行向量相似度搜索 关键词 BM25 搜索再用RRFReciprocal Rank Fusion加权融合结果。LangChain 的EnsembleRetriever可一行代码启用“EnsembleRetriever(retrievers[vector_retriever, bm25_retriever], weights[0.7, 0.3])”。实测在法律条文检索中RRF 比纯向量检索将 top-1 准确率从 61% 提升至 79%。第四层重排序Reranking——用小模型拯救大模型召回的 Top-20 chunk 里真正相关的可能只有 2-3 个。直接喂给 LLM 会稀释注意力。解决方案是引入轻量级重排序模型如BAAI/bge-reranker-base。它只有 125M 参数CPU 上推理延迟 200ms。调用方式极简reranker.rank(query, [chunk1, chunk2, ...])返回按相关性重排的列表。注意其输入格式要求 query 和 chunk 必须拼接为query: {q} passage: {p}漏掉前缀会导致效果归零。第五层提示词工程Prompt Engineering——不是写得越长越好一个反直觉的事实在 RAG 场景下提示词越简洁LLM 越不容易“编造”。我的黄金模板是三段式①角色定义10 字内“你是一名资深税务顾问”②任务指令20 字内“仅根据提供的材料回答不确定则说‘无法确定’”③上下文注入严格限定“【材料开始】{retrieved_chunks}【材料结束】”。绝对避免“请结合你的知识和以下材料”这类开放性表述——这等于授权 LLM 自由发挥。实测显示删除“结合你的知识”五个字幻觉率下降 52%。注意RAG 的终极瓶颈往往不在模型而在数据新鲜度。我们曾用 Airflow 每日凌晨执行git pull更新知识库源文件触发自动 re-embedding。但发现 PDF 文件的mtime不随内容更新导致增量更新失效。最终方案是在源文件目录下生成checksums.json记录每个文件的 md5每次只 re-embed checksum 变化的文件。3.2 Agents 的本质是“可控的失控”其核心在于状态管理与工具契约如果说 RAG 是“增强回答”那么 Agents 就是“执行任务”。但很多项目把 Agent 做成了“高级版 Chatbot”这是根本性误判。真正的 Agent 必须满足三个刚性条件目标导向Goal-Oriented、状态感知State-Aware、工具自治Tool-Autonomous。三者缺一不可。目标导向用结构化 Goal 替代模糊 Prompt不要写“帮用户订机票”而要定义{goal: book_flight, constraints: [departure_city上海, arrival_city北京, date2024-06-15, max_price1200], required_outputs: [flight_number, departure_time, price]}。LangChain 的AgentExecutor支持tool_names白名单但更关键的是在Agent初始化时传入return_intermediate_stepsTrue这样你能看到每一步的action_input是否符合预期。我曾在一个电商客服 Agent 中发现当用户说“我想退上个月买的耳机”LLM 生成的action_input是{order_id: last_month_headphones}—— 这根本不是有效订单 ID。解决方案是在工具调用前插入一个ValidationTool用正则校验order_id格式不合法则强制要求 LLM 重新生成。状态感知Agent 的“记忆”不是 LLM 的上下文窗口把所有历史对话塞进system_prompt是灾难。正确做法是分层存储①短期记忆Short-term用ConversationBufferWindowMemory保留最近 5 轮对话避免重复提问②长期记忆Long-term将用户 profile、偏好、历史订单存入 Rediskey 为user:{id}:profile③任务状态Task-state为每个进行中的任务生成唯一task_id状态存入 PostgreSQL 的task_state表字段包括current_step,tool_inputs,retry_count。当 Agent 因网络超时中断重启后可通过task_id恢复到step3而不是从头开始。工具自治定义清晰的 Tool Contract 是成败关键每个工具必须有机器可读的契约Contractclass SearchFlightTool(BaseTool): name search_flight description Search flights by departure/arrival city and date. Returns list of flights with flight_number, departure_time, price. args_schema: Type[BaseModel] FlightSearchInput # Pydantic model defining exact input fields def _run(self, departure_city: str, arrival_city: str, date: str) - List[Dict]: # 实际调用航班 API pass关键点在于args_schema它强制 LLM 在生成action_input时必须输出符合FlightSearchInput结构的 JSON。如果 LLM 输出{from: shanghai}args_schema会抛出 ValidationError触发 Agent 自动修正。这比任何提示词约束都可靠。我们曾用此机制将工具调用失败率从 34% 降至 1.2%。4. 实操过程与核心环节实现从零搭建一个可商用的 RAGAgent 混合系统4.1 环境准备与依赖锁定用 Poetry 而非 Pip 的深层原因选择 Poetry 而非 piprequirements.txt核心原因是可重现性Reproducibility。pip 安装时会自动升级次版本号如langchain0.1.0可能安装0.1.15而不同 patch 版本间常有不兼容变更。Poetry 的poetry.lock文件精确锁定每个包的完整哈希值确保poetry install在任何机器上安装完全一致的依赖树。初始化步骤# 创建项目 poetry init -n --name rag-agent-prod --description Production-ready RAGAgent system # 添加核心依赖注意版本锁死 poetry add langchain0.1.16 \ langchain-community0.0.36 \ llama-index0.10.32 \ pymilvus2.4.5 \ openai1.30.5 \ fastapi0.111.0 \ uvicorn0.29.0 \ redis4.6.0 \ sqlalchemy2.0.30 # 生成 lock 文件关键 poetry lock # 导出为 requirements.txt 供 CI 使用可选 poetry export -f requirements.txt --without-hashes requirements.txt实操心得在pyproject.toml中显式声明 Python 版本约束[tool.poetry.dependencies] python ^3.11。我们曾因某次 CI 环境默认使用 Python 3.12导致pymilvus的 C 扩展编译失败排查耗时 8 小时。锁定 Python 版本是底线。4.2 构建可审计的知识库管道从 PDF 到向量的全链路以公司内部《2024 产品白皮书.pdf》为例构建生产级 RAG 管道Step 1PDF 解析与结构化清洗不用PyPDF2改用unstructured的partition_pdffrom unstructured.partition.pdf import partition_pdf elements partition_pdf( filename2024_product_whitepaper.pdf, strategyhi_res, # 高精度模式识别表格和公式 infer_table_structureTrue, include_page_breaksFalse, languages[zh], # 关键指定坐标系避免 OCR 错位 coordinatesTrue )elements是一个List[Element]每个Element包含text,categoryTitle, NarrativeText, Table,metadata页码、坐标。这比纯文本提取多出 3 倍元数据为后续分块提供依据。Step 2语义分块与元数据注入用llama-index的SentenceSplitter但改造其逻辑from llama_index.core.node_parser import SentenceSplitter # 自定义分块器优先保持段落完整再按句子切 splitter SentenceSplitter( chunk_size256, chunk_overlap32, paragraph_separator\n\n, # 以双换行分段 secondary_chunking_regex[^。][。]?, # 中文句子切分正则 ) nodes splitter.get_nodes_from_documents([Document(textcleaned_text, metadata{source: whitepaper_v2024})]) # 为每个 node 注入章节标题从 elements 中提取 for node in nodes: if section_title in node.metadata: node.text f【{node.metadata[section_title]}】{node.text}Step 3向量化与 Milvus 索引构建使用bge-m3模型支持中英混合检索from sentence_transformers import SentenceTransformer from pymilvus import connections, Collection, FieldSchema, DataType, CollectionSchema # 连接 Milvus connections.connect(default, hostmilvus-etcd, port2379) # 定义 Schema fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(namevector, dtypeDataType.FLOAT_VECTOR, dim1024), FieldSchema(nametext, dtypeDataType.VARCHAR, max_length65535), FieldSchema(namesource, dtypeDataType.VARCHAR, max_length256), FieldSchema(namepage, dtypeDataType.INT32), ] schema CollectionSchema(fields, rag_collection) # 创建 Collection collection Collection(rag_collection, schema) collection.create_index( field_namevector, index_params{ index_type: HNSW, metric_type: COSINE, params: {M: 16, efConstruction: 128} } ) # 批量插入关键batch size128避免内存溢出 model SentenceTransformer(BAAI/bge-m3) vectors model.encode([node.text for node in nodes], batch_size128) entities [ vectors.tolist(), [node.text for node in nodes], [node.metadata.get(source, ) for node in nodes], [node.metadata.get(page, 0) for node in nodes], ] collection.insert(entities) collection.flush()Step 4混合检索与重排序服务封装构建 FastAPI 端点整合向量检索、BM25、RRF、重排序from fastapi import FastAPI, HTTPException from rank_bm25 import BM25Okapi import numpy as np app FastAPI() # 预加载 BM25 语料从 Milvus 中拉取所有 text 字段 bm25_corpus [row[text] for row in collection.query(exprid 0, output_fields[text])] tokenized_corpus [doc.split() for doc in bm25_corpus] bm25 BM25Okapi(tokenized_corpus) app.post(/rag_search) def rag_search(query: str): # Step 1: 向量检索 query_vector model.encode([query])[0] res collection.search( data[query_vector], anns_fieldvector, param{metric_type: COSINE, params: {ef: 128}}, limit20, output_fields[text, source, page] ) # Step 2: BM25 检索 tokenized_query query.split() bm25_scores bm25.get_scores(tokenized_query) bm25_top_k np.argsort(bm25_scores)[::-1][:20] # Step 3: RRF 融合 rrf_scores {} for i, (id, score) in enumerate(res[0]): rrf_scores[id] 1 / (i 60) # 向量检索 RRF 权重 for i, idx in enumerate(bm25_top_k): doc_id idx # 简化实际需映射 rrf_scores[doc_id] rrf_scores.get(doc_id, 0) 1 / (i 60) # Step 4: 重排序调用 bge-reranker reranked reranker.rank(query, [corpus[i] for i in list(rrf_scores.keys())[:10]]) return {results: [{text: r.text, score: r.score} for r in reranked]}4.3 构建状态感知的 Agent 工作流以“智能报销审核”为例场景员工上传发票图片Agent 需完成 OCR → 提取金额/日期/商户 → 查询差旅政策 → 判断是否超标 → 生成审核意见。Step 1定义工具契约Tool Contractfrom pydantic import BaseModel, Field from typing import List, Dict, Optional class InvoiceOCRInput(BaseModel): image_url: str Field(..., descriptionInvoice image URL, must be accessible) class PolicyCheckInput(BaseModel): expense_type: str Field(..., descriptione.g., airfare, hotel) amount: float Field(..., descriptionExpense amount in CNY) date: str Field(..., descriptionISO format date, e.g., 2024-06-15) class InvoiceOCRTool(BaseTool): name ocr_invoice description Extract text from invoice image. Returns {amount: float, date: str, merchant: str}. args_schema: Type[BaseModel] InvoiceOCRInput def _run(self, image_url: str) - Dict: # 调用 PaddleOCR API return {amount: 1280.0, date: 2024-06-10, merchant: 上海虹桥机场} class PolicyCheckTool(BaseTool): name check_policy description Check if expense complies with company policy. Returns {compliant: bool, reason: str}. args_schema: Type[BaseModel] PolicyCheckInput def _run(self, expense_type: str, amount: float, date: str) - Dict: # 查询 PostgreSQL 政策表 return {compliant: False, reason: Airfare exceeds 1200 CNY limit}Step 2构建状态感知 Agent Executorfrom langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 提示词模板强调状态与约束 prompt ChatPromptTemplate.from_messages([ (system, You are an AI finance assistant. You MUST follow these rules: 1) Only use provided tools. 2) If a tool fails, report error and stop. 3) Output final answer in Chinese, with clear 通过 or 不通过结论.), (placeholder, {chat_history}), (human, {input}), (placeholder, {agent_scratchpad}), ]) # 初始化 LLM使用本地 Ollama 模型降低延迟 llm ChatOpenAI( base_urlhttp://localhost:11434/v1, api_keyollama, modelqwen2:7b, temperature0.1 ) # 创建 Agent agent create_tool_calling_agent(llm, [InvoiceOCRTool(), PolicyCheckTool()], prompt) # 状态感知的 Executor关键注入 task_id class StatefulAgentExecutor(AgentExecutor): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.task_state_db TaskStateDB() # 自定义状态数据库 def invoke(self, input: Dict, config: Optional[Dict] None) - Dict: task_id input.get(task_id, str(uuid.uuid4())) # 从 DB 恢复状态如果存在 state self.task_state_db.get(task_id) if state and state.current_step ocr_invoice: # 跳过 OCR直接用缓存结果 input[ocr_result] state.ocr_result result super().invoke(input, config) # 保存状态 self.task_state_db.save(task_id, { current_step: result.get(intermediate_steps, [])[-1][0].tool if result.get(intermediate_steps) else done, ocr_result: result.get(ocr_result) }) return result agent_executor StatefulAgentExecutor(agentagent, tools[InvoiceOCRTool(), PolicyCheckTool()])Step 3暴露为 REST APIapp.post(/submit_invoice) def submit_invoice(image_url: str): task_id str(uuid.uuid4()) try: result agent_executor.invoke({ input: f审核这张发票{image_url}, task_id: task_id, chat_history: [] }) return {task_id: task_id, result: result[output]} except Exception as e: # 记录详细错误日志含 task_id logger.error(fAgent execution failed for task {task_id}: {str(e)}) raise HTTPException(status_code500, detail审核失败请重试)5. 常见问题与排查技巧实录那些文档里绝不会写的血泪教训5.1 RAG 场景高频问题速查表问题现象根本原因排查步骤解决方案召回内容与问题无关向量模型未针对中文微调① 用bge-m3替换all-MiniLM-L6-v2② 检查 embedding 时是否添加了query:前缀model.encode([fquery: {q}])同一问题多次提问答案不一致缓存了旧的向量索引① 查milvuscollection 的created_timestamp② 检查 re-embedding 脚本是否真的执行在 re-embedding 脚本末尾添加collection.flush()长文档回答遗漏关键信息分块时切碎了跨页表格① 用unstructured的strategyhi_res② 检查elements中categoryTable的数量对Table类型元素用table_to_html()转为字符串再切块API 响应延迟忽高忽低MilvusHNSW的ef参数过大① 查milvus日志中的search耗时② 用explain命令分析查询计划将ef从 512 降至 128牺牲 0.3% recall 换取 60% latency 下降实操心得当遇到“召回率低”问题第一反应不是换模型而是检查PDF 解析质量。用unstructured解析后随机抽 10 页人工核对elements中的text是否与原文一致。我们曾发现某版unstructured对扫描版 PDF 的 OCR 识别率仅 42%切换到pymupdfpaddleocr组合后提升至 91%。5.2 Agents 场景致命陷阱与绕过方案陷阱一“工具调用死循环”现象Agent 反复调用同一个工具如search_flight输入参数几乎不变。原因LLM 未理解工具返回结果或args_schema定义过于宽松如date: str未约束格式。绕过方案在工具_run方法中加入强校验def _run(self, departure_city: str, arrival_city: str, date: str) - List[Dict]: # 强制日期格式校验 try: datetime.strptime(date, %Y-%m-%d) except ValueError: raise ValueError(fInvalid date format: {date}, expected YYYY-MM-DD) # 实际调用 return call_flight_api(...)陷阱二“状态丢失导致任务中断”现象用户在多轮对话中说“继续刚才的报销”Agent 却从头开始。原因ConversationBufferMemory未持久化或task_id未在前后端传递。绕过方案前端必须管理 task_id。在首次请求时API 返回{task_id: xxx, session_id: yyy}后续请求必须携带session_id后端用session_id查 Redis 获取task_id。我们用fastapi-session库实现配置backendRedisBackend(redis_client)。陷阱三“LLM 拒绝调用工具”现象明明提供了search_flight工具但 LLM 总是生成Final Answer。原因提示词中description过于简略或tool_names未在system_prompt中显式列出。绕过方案在ChatPromptTemplate的system消息中硬编码工具列表(system, Available tools: ocr_invoice, check_policy. You MUST use one of them when needed.)5.3 性能压测与容量规划别让“高并发”成为上线噩梦一个常被忽视的事实RAGAgent 系统的瓶颈永远不在 LLM 推理而在 I/O 和向量检索。我们对一个 50 万 chunk 的知识库做了压测并发数平均延迟错误率瓶颈定位优化方案10850ms0%Milvus CPU 使用率 75%增加 Milvusquerynode副本数502.1s12%Redis 连接池耗尽将redis-py连接池max_connections1001005.3s47%bge-m3embedding GPU 显存不足改用bge-small-zh-v1.5显存占用降 60%关键经验永远假设 LLM 是黑盒只优化你能控制的部分。我们最终的生产配置是Milvus3 个querynode8C16GHNSW索引ef64Redis1 主 2 从连接池max_connections200Embedding 模型bge-small-zh-v1.5CPU 推理延迟 300msLLMOllamaqwen2:7bGPU 推理num_ctx4096最后分享一个小技巧在 FastAPI 中用app.middleware(http)记录每个请求的task_id、start_time、end_time、status_code写入 Elasticsearch。当线上报警时直接 Kibana 查task_id就能看到完整调用链OCR 耗时 1200ms → Policy Check 耗时 80ms → LLM 生成耗时 450ms → 总耗时 1730ms。这比任何日志 grep 都高效。