知识广场

不想把数据发到 OpenAI 但想用大模型？Ollama 让你本地跑 Llama 3 / Qwen / DeepSeek / Mistral 等开源模型，LangChain 提供统一封装做 RAG / agent。 整套零成本（GPU 电费除外），数据完全在本地。 ## 1. 装 Ollama ```bash curl -fsSL https://ollama.com/install.sh | sh # macOS: brew install ollama # Windows: 从 ollama.com 下安装包 ollama --version ``` Ollama 作为后台服务跑（端口 11434）： ```bash systemctl start ollama # Linux brew services start ollama # macOS ``` ## 2. 拉个模型 ```bash ollama pull qwen2.5:7b # 或更小的：qwen2.5:3b （4 GB 显存就能跑） # 或更大的：qwen2.5:14b、llama3.1:70b（需要 24+ GB / 80 GB） ollama list ``` ## 3. 命令行直接聊 ```bash ollama run qwen2.5:7b >>> 用 Python 写一个二分查找 ``` `Ctrl-D` 退出。 ## 4. HTTP API（OpenAI 兼容） Ollama 默认在 `:11434` 暴露 OpenAI 兼容 endpoint： ```bash curl http://localhost:11434/v1/chat/completions \ -H "Content-Type: application/json" \ -d '{ "model": "qwen2.5:7b", "messages": [{"role": "user", "content": "你好"}] }' ``` 任何 OpenAI 库直接换 base URL 就能用： ```python from openai import OpenAI client = OpenAI(base_url='http://localhost:11434/v1', api_key='ollama') resp = client.chat.completions.create( model='qwen2.5:7b', messages=[{'role': 'user', 'content': '你好'}], ) print(resp.choices[0].message.content) ``` ## 5. LangChain 集成 ```bash uv add langchain langchain-community langchain-ollama ``` ```python from langchain_ollama import ChatOllama, OllamaEmbeddings llm = ChatOllama(model='qwen2.5:7b', temperature=0.3) emb = OllamaEmbeddings(model='nomic-embed-text') # 嵌入模型，需另拉 ``` `ollama pull nomic-embed-text` 拉一个 274MB 的嵌入模型。 ## 6. RAG：用本地 LLM + 本地知识库回答 ```python from langchain_chroma import Chroma from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain.chains import RetrievalQA from langchain_community.document_loaders import TextLoader # 1. 加载文档（这里举例一个 txt，实际可能是 md / pdf / html） loader = TextLoader('knowledge.md', encoding='utf-8') docs = loader.load() # 2. 切块 splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50) chunks = splitter.split_documents(docs) # 3. 向量化 + 入库 vectorstore = Chroma.from_documents(chunks, embedding=emb, persist_directory='./chroma_db') # 4. 检索 + LLM 回答 qa = RetrievalQA.from_chain_type( llm=llm, chain_type='stuff', retriever=vectorstore.as_retriever(search_kwargs={'k': 4}), return_source_documents=True, ) result = qa.invoke({'query': '本文如何解释 X 概念？'}) print(result['result']) print('---sources---') for d in result['source_documents']: print(d.page_content[:80]) ``` 整个 pipeline 完全在本地。 ## 7. 持久化向量库 ```python # 写入后端 vectorstore.persist() # Chroma 自动 persist 到 directory # 之后加载已有库 vectorstore = Chroma(persist_directory='./chroma_db', embedding_function=emb) ``` Chroma 是文件型向量数据库，适合 < 100 万 chunk 的小规模 RAG。 大规模用 Qdrant / Milvus / Weaviate。 ## 8. Agent / tools ```python from langchain.agents import create_react_agent, AgentExecutor from langchain.tools import tool from langchain import hub @tool def get_weather(city: str) -> str: """获取指定城市的当前天气。""" # 假装调 API return f'{city} 今天晴，22°C' @tool def calculator(expr: str) -> str: """计算数学表达式，例如 '2 * (3 + 4)'。""" try: return str(eval(expr, {'__builtins__': {}})) except Exception as e: return f'error: {e}' prompt = hub.pull('hwchase17/react') agent = create_react_agent(llm, [get_weather, calculator], prompt) executor = AgentExecutor(agent=agent, tools=[get_weather, calculator], verbose=True, max_iterations=4) executor.invoke({'input': '北京天气怎么样？顺便算一下 47 * 12'}) ``` `verbose=True` 输出 agent 的思考过程（很好玩，但生产关掉）。 ## 9. 流式输出 ```python for chunk in llm.stream('用 100 字介绍 RAG'): print(chunk.content, end='', flush=True) ``` 或者在 FastAPI 里返回 SSE 流。 ## 10. 性能 tip - **量化**：`ollama pull qwen2.5:7b-instruct-q4_K_M` 4-bit 量化， 4 GB 显存 / 内存就能跑。精度降几个点 - **多模型切换**：`ollama list` + `ollama run`，自动 load / unload - **并发**：Ollama 默认串行处理，需要并发的话调 `OLLAMA_NUM_PARALLEL=4` - **GPU**：自动检测 CUDA / Metal；CPU 也能跑但慢 5-10x - **保持模型在内存**：`ollama keep-alive` 控制；默认 5 分钟没请求会卸载 ## 11. 模型选择 | 用途 | 推荐 | 显存 | |---|---|---| | 通用对话 / RAG | qwen2.5:7b / 14b | 8 / 16 GB | | 编程 | qwen2.5-coder:7b / deepseek-coder:6.7b | 8 GB | | 中文为主 | qwen2.5、yi、deepseek | - | | 极轻量 | qwen2.5:3b、phi3:mini | 4 GB | | 顶配 | llama3.1:70b、qwen2.5:72b | 48-80 GB | | 嵌入 | nomic-embed-text、bge-m3 | < 2 GB | ## 12. 数据安全 - Ollama 默认监听 `127.0.0.1:11434`，不暴露公网 - LangChain 调用的所有 API 都走本地 - 向量库（Chroma）默认本地文件 - 整个 pipeline 不发任何数据到云端 完美的 enterprise / 隐私敏感场景。 ## 踩过的坑 - 第一次拉大模型很慢（GB 级），可以预先 `ollama pull` 而不是等首次调用 超时。 - 7B 模型在 CPU 上跑非常慢（每 token ~1 秒），交互式不可用；GPU / Apple Silicon 必备。 - 上下文窗口默认很小（2048）。Ollama Modelfile 可以加大： ``` FROM qwen2.5:7b PARAMETER num_ctx 8192 ``` `ollama create my-qwen -f Modelfile`。 - LangChain 升级特别快，API 经常变。在 pyproject.toml lock 版本， 不要随便升。

## 起因 第一次做 RAG 时按"每 500 字符一刀"切了 1 万份文档进向量库， 召回出来的 chunk 经常是半句话开头、半句话结尾。LLM 看着这种残缺片段 回答经常张冠李戴："根据文档 ...（被截断）"。 解决回答质量问题，70% 在切块上下功夫，30% 在检索算法。 ## 三种切块策略 ### A. 固定长度（最简单） ```python def chunk_fixed(text, size=500, overlap=50): chunks = [] for i in range(0, len(text), size - overlap): chunks.append(text[i:i+size]) return chunks ``` `overlap` 让相邻 chunk 有 50 字符重叠，避免句子被切两半时只有半句进 任一 chunk。优点：实现简单，长度均匀；缺点：仍可能在标点 / 段落中间截断。 ### B. 语义边界（推荐） 按段落、句子、Markdown 标题切。LangChain 的 `RecursiveCharacterTextSplitter` 最常用： ```python from langchain_text_splitters import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, separators=["\n\n", "\n", "。", "！", "？", ". ", " ", ""], ) chunks = splitter.split_text(text) ``` `separators` 顺序很重要：先按双换行切（段落），不够再按单换行（行）， 还不够按句号，最后才硬切。中文要加 `"。"`、`"！"`、`"？"`， 英文版本默认有 `"."`。 代码 / 表格 / Markdown 文档： ```python from langchain_text_splitters import MarkdownHeaderTextSplitter splitter = MarkdownHeaderTextSplitter( headers_to_split_on=[("#", "h1"), ("##", "h2"), ("###", "h3")], ) chunks = splitter.split_text(md_text) # 每 chunk 自动带 metadata { h1: "...", h2: "..." } ``` 按 Markdown 标题切的好处：chunk 自带"它属于哪一节"的元数据，可以在 prompt 里附带给 LLM 当上下文。 ### C. 父子层级（parent-child retrieval） 切两份： - **小 chunk**（200-300 字符）用于 embedding + 检索（细颗粒，语义匹配准） - **大 chunk**（1000-2000 字符）用于喂给 LLM（提供完整上下文） 存储时小 chunk 通过 metadata 指向所属大 chunk： ```python from langchain.retrievers import ParentDocumentRetriever from langchain.storage import InMemoryStore parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000) child_splitter = RecursiveCharacterTextSplitter(chunk_size=400) retriever = ParentDocumentRetriever( vectorstore=Chroma(embedding=emb), docstore=InMemoryStore(), child_splitter=child_splitter, parent_splitter=parent_splitter, ) retriever.add_documents(docs) # 查询时：小 chunk 匹配 → 返回对应大 chunk chunks = retriever.invoke('问题') ``` ## 效果对比 我在一个内部知识库（5000 篇文档）做过 A/B： | 策略 | 检索准确率 | LLM 回答完整度 | 实现复杂度 | |---|---|---|---| | 固定长度 | 62% | 70%（多被截断） | 极简 | | 语义边界 | 78% | 88% | 简单 | | 父子层级 | 81% | 94% | 中 | **结论**：起步用语义边界（RecursiveCharacterTextSplitter），效果不够好 再升级到父子。固定长度只在快速 POC 时用。 ## 其它要诀 ### 1. chunk size 不是越小越好 300 字符的 chunk 召回时上下文太少；2000 字符的 chunk embedding 时 语义被"稀释"，匹配不准。**400-800 字符是甜点**（英文）/ **200-400 中文字符** （中文每字承载语义比英文 token 多）。 ### 2. metadata 一起存 ```python chunk = { 'text': '...', 'metadata': { 'source': 'manual.md', 'section': 'Installation', 'date': '2024-01-15', } } ``` 检索时可以 filter（只查"最近 30 天的内容"），rerank 时可以加权。 ### 3. 长文档加 summary chunk 文档开头加一个"全文摘要" chunk（用 LLM 生成）作为 top-level entry。 检索时 summary chunk 匹配 → 暗示整篇文档相关 → 扩展取相邻 chunk。 ## 踩过的坑 1. **Tokenizer 不匹配**：embedding model 是按 token 计长，但我按字符切 500，可能实际是 800-1200 token，超过 model 上限被截。改成 `from_huggingface_tokenizer` 按真实 token 切： ```python splitter = RecursiveCharacterTextSplitter.from_huggingface_tokenizer( tokenizer, chunk_size=200, chunk_overlap=20) ``` 2. **PDF 提取文字时表格变乱**：纯文本提取（pdfplumber）保留不了表格结构。 表格密集的 PDF 用 `unstructured` 库专门处理表格，或者 PDF → HTML → 切。 3. **重复内容污染检索**：每篇文档都有相同的 "本文档版权所有…" 页脚 → embedding 集中在这个 footer，检索结果偏向有 footer 的文档。 切之前先 strip 这些 boilerplate。 4. **多语言文档**：中英文混排时 RecursiveCharacterTextSplitter 默认 separators 不包含中文标点。手动加 `"。"`、`"！"`、`"？"`、`"；"`。

## 起因 写了一个 LangChain agent 帮用户查数据库 + 写 SQL + 解释结果， 跑起来时不时给出乱七八糟的答案。问题可能出在： - 哪一步的 prompt 让 LLM 跑偏？ - 调了什么 tool、tool 返回了什么？ - 重试了几次？ - 哪段花了最多 token / 最长时间？ `print(intermediate_steps)` 看不出来。LangSmith 是 LangChain 团队的可观测平台， 自动把 chain / agent 的每一次执行都记下来，UI 时间线展开看。 ## 解决方案 ### 注册 + 装 注册 [smith.langchain.com](https://smith.langchain.com)（免费层个人项目够）， 拿 API key。 ```bash uv add langsmith export LANGCHAIN_TRACING_V2=true export LANGCHAIN_API_KEY=ls__xxxxxxxx export LANGCHAIN_PROJECT=my-agent-debug ``` **就这样**。LangChain 自动把所有 chain / agent 执行 trace 上报到 LangSmith。 不需要改任何代码。 ### 看 trace 进 LangSmith UI → 选 project → 看 trace 列表： ``` Run name Duration Tokens Status sql_agent.invoke 5.3s 2,341 success sql_agent.invoke 12.1s 8,902 error ... ``` 点进任意 run： ``` └─ sql_agent (5.3s, 2341 tokens) ├─ planner (1.2s, 540 tokens) │ └─ ChatOpenAI (1.1s, 540 tokens) │ ▶ prompt: "You are a SQL planner. Decide..." │ ◀ output: {"action": "run_sql", "sql": "SELECT ..."} ├─ tool: run_sql (0.3s) │ ▶ input: SELECT count(*) FROM users WHERE ... │ ◀ output: [{"count": 142}] └─ responder (3.8s, 1801 tokens) └─ ChatOpenAI (3.7s, 1801 tokens) ▶ prompt: "Given the query result, answer..." ◀ output: "总共有 142 个符合条件的用户。" ``` 每一层展开看完整 prompt + completion + tokens + latency。 ### 找出"为什么这次出错" filter "status=error" 看所有失败 run。点进去： ``` └─ sql_agent (failed at responder) ├─ planner ✓ ├─ run_sql ✗ (ERROR: column "user_typ" does not exist) └─ ... ``` 清楚看到 SQL agent 拼错列名（应该是 user_type）。回头改 prompt 加 schema 提示。 ### A/B 对比 prompt LangSmith 有 "Datasets" + "Compare experiments"： 1. 创建 dataset：10-20 个典型 query + 期望答案 2. 跑 prompt v1：`run_on_dataset(dataset_name, prompt_v1_chain)` 3. 跑 prompt v2：同上 4. UI 对比每个 input 上 v1 vs v2 的输出 + 自动 eval 分数 ```python from langsmith import Client client = Client() dataset = client.create_dataset('sql_agent_eval') client.create_example( inputs={'query': '过去 7 天注册的用户数'}, outputs={'expected': '~150'}, dataset_id=dataset.id, ) from langchain.smith import RunEvalConfig client.run_on_dataset( dataset_name='sql_agent_eval', llm_or_chain_factory=lambda: sql_agent_v2, evaluation=RunEvalConfig(evaluators=['qa', 'context_qa']), ) ``` LangSmith 自动用 GPT-4 当 judge 评分。 ### prompt hub LangChain Hub 集成在 LangSmith： ```python from langchain import hub prompt = hub.pull('rlm/rag-prompt') ``` 社区 prompt 模板，pull + 改自己版本 + push 回去（你的私有 namespace）。 ### 用于 production 监控 不止 dev 用： ```python import os os.environ['LANGCHAIN_TRACING_V2'] = 'true' os.environ['LANGCHAIN_PROJECT'] = 'prod' # 区分环境 ``` production 上跑的每个 trace 都收集。看： - 每天调用量 - 每个 chain 的 P50 / P95 延迟 - token 消耗趋势 - error 率 + 错误类型分布 可以接 alert：当 error rate > 5% 邮件通知。 ## 与替代品对比 | | LangSmith | Langfuse | Phoenix (Arize) | |---|---|---|---| | 开源 / 自托管 | ❌（cloud 为主，自托管要 enterprise） | ✅ 全开源 | ✅ | | 与 LangChain 集成 | 最原生 | 中 | 中 | | 与 LlamaIndex | 中 | 中 | 强 | | eval 框架 | ✅ | ✅ | ✅ | | 价格 | 免费 5k traces/月 | 完全免费（自托管） | 免费层 | LangChain 项目用 LangSmith 最方便；不想绑定平台用 Langfuse 自托管。 ## 效果 - agent 失败率 12% → 4%，靠看 trace 改 prompt 一周搞定 - "为什么这次跑出 X" 的问题从"猜 + 加 print"变成 "去 LangSmith 看一下" - 找到一个 prompt 让 token 消耗减半（trace 里看到 LLM 反复重复同样 上下文） - 团队 review prompt 改动有了客观依据（运行某 dataset 对比 v1/v2 分数） ## 踩过的坑 1. **traces 含敏感数据上 cloud**：用户邮箱 / 手机号都进 prompt 时 隐私问题。开 `LANGCHAIN_TRACING_V2=false` 临时关，或者 enterprise 自托管。 2. **大批量 run 上报慢**：默认同步上报，每个 run 加 50-100ms。设 `LANGCHAIN_CALLBACKS_BACKGROUND=true` 异步上报。 3. **trace 嵌套太深**：复杂 agent 数十层调用，UI 加载慢。用 tag / metadata 标记关键步骤再筛选。 4. **eval 用 GPT-4 评分**：成本可能比被评模型还高。先小 dataset 验证 eval 设置对，再扩规模。 5. **本地 LLM trace**：用 Ollama / vLLM 跑本地模型时，把 ChatOpenAI 的 base_url 改本地即可，trace 一样上报。注意 token 计数对本地 模型不准。

## 起因 接了一个"自动分类客服工单" 的任务。最 naive prompt： ``` 分类下面的客服请求到一个类别： {ticket} ``` 效果：60% 正确率 + 经常输出无关解释 + 偶尔编造类别。 调了几轮 prompt 后到 92%，token 使用减半。下面是几个真正提分的 pattern，不是"魔法咒语"。 ## Pattern 1: 明确角色 + 任务 + 输出格式 ``` 你是客服工单分类专家。 任务：把客户请求归到下面一个类别。 类别（严格选其中一个）： - billing (付款 / 发票 / 退款) - bug (产品故障 / 异常报错) - feature_request (希望新增功能) - account (账号登录 / 权限 / 密码) - other (上面都不对) 输出格式： {"category": "<类别名>", "confidence": <0-1>} 不要输出任何额外解释。 客户请求： {ticket} ``` 为什么有效： - **角色**：把模型从"通用助手" 引导到"分类专家"语境 - **限制类别**：明确告诉它可选范围，减少幻觉 - **JSON 输出**：好解析 + 没散文废话 - **"不要输出额外解释"**：明确禁止前/后缀闲话 立刻从 60% → 80%。 ## Pattern 2: Few-shot 例子 光描述类别不够，给具体例子： ``` [上面的 prompt...] 例子： 输入：「我的卡被扣了两次，请退一次」 输出：{"category": "billing", "confidence": 0.95} 输入：「点击保存按钮后页面卡死」 输出：{"category": "bug", "confidence": 0.9} 输入：「能不能加个深色模式？」 输出：{"category": "feature_request", "confidence": 0.95} 输入：{ticket} 输出： ``` 3-5 个例子覆盖"边界 / 容易混淆" 的情况效果最好。 "为什么这个分到 billing 而不是 account" 类的边界 case 用例子讲清楚。 → 80% → 90%。 ## Pattern 3: Chain of thought（思考过程） 对推理类任务（不是简单分类）让模型先"想"再答： ``` 你是数学题解题专家。 题目：一个商店周一卖了 23 件商品，周二是周一的 1.5 倍， 周三是周一周二之和的一半。三天总共卖了多少？ 要求： 1. 先逐步推理（标 [思考]） 2. 再给最终答案（标 [答案]） ``` 模型输出： ``` [思考] - 周一：23 件 - 周二：23 × 1.5 = 34.5 件 → 取整 35（实际 35 才合理） - 周三：(23 + 35) / 2 = 29 件 - 总：23 + 35 + 29 = 87 件 [答案] 87 ``` 直接问"答案是？"很多时候算错。让它写推理过程后准确率显著上升。 适合：数学 / 推理 / 多步逻辑。 不适合：简单 lookup / 分类（CoT 浪费 token）。 ## Pattern 4: Structured output (JSON mode / function call) 不要让 LLM 用 markdown 包 JSON 后让你正则提取——浪费 token + 不稳定。 直接用 API 的 structured output： ```python from openai import OpenAI import json client = OpenAI() resp = client.chat.completions.create( model='gpt-4o', messages=[{'role': 'user', 'content': prompt}], response_format={ 'type': 'json_schema', 'json_schema': { 'name': 'ticket_classification', 'schema': { 'type': 'object', 'properties': { 'category': { 'type': 'string', 'enum': ['billing', 'bug', 'feature_request', 'account', 'other'], }, 'confidence': {'type': 'number', 'minimum': 0, 'maximum': 1}, }, 'required': ['category', 'confidence'], 'additionalProperties': False, }, 'strict': True, } } ) result = json.loads(resp.choices[0].message.content) ``` `strict: true` 让模型在 decoding 时只产生 schema 合法 token。 **100% 合规 JSON**，从此不需要 try/except parse。 Anthropic 用 tool use / Gemini 用 response_schema 同理。 ## Pattern 5: 减少 token = 省钱 + 快 token 不仅是钱，还是延迟。每个 token 大模型 50-200ms。 技巧： 1. **删冗余形容词**："请帮忙仔细认真地分析下面..." → "分析：" 2. **缩短例子**：3 个例子够时不放 10 个 3. **不重复 system + user**：system 里写完类别后 user 不重复 4. **truncate 长输入**：保留 ticket 前 1000 字符（多数信号在开头） 5. **batch 处理**：5 个工单一起喂（"分析下面 5 个 ticket"）省 system token ```python # 单条 vs batch single_cost = (system_tokens + ticket_tokens + output) * N batch_cost = (system_tokens + N * ticket_tokens + N * output) # batch 省的是 N 份 system tokens ``` 我们 batch=10 后 token 量降 30%。注意 batch 太大模型 attention 分散 精度反而降，5-15 是甜点。 ## Pattern 6: 自检 / verifier loop 对关键任务，跑两次让另一个 prompt 验证： ```python # Pass 1: 主分类 result = classify(ticket) # Pass 2: 验证 verify_prompt = f""" 有人把这个工单分类为 '{result['category']}'。是否合理？ 工单：{ticket} 输出 JSON: {{"agree": true|false, "reason": "..."}} """ verify = llm(verify_prompt) if not verify['agree']: # 退到人工 review 队列 flag_for_human(ticket, result, verify['reason']) ``` cost 翻倍但低置信度 ticket 被人工接管，整体准确率上 96%+。 ## 一些反 pattern ### ❌ 写一堆"必须 / 不要 / 绝对" ``` 你必须按下面规则。 你绝对不能输出 X。 你一定要返回 JSON。 你不允许加任何解释。 你必须用中文。 ``` 模型对负面指令响应一般。改成正面： ``` 输出只包含 JSON。返回的语言是中文。 ``` ### ❌ 提供模糊定义 ``` 分类为： - 重要的问题 - 不重要的问题 ``` "重要" 没定义 → 模型自己猜 → 不稳定。 给具体标准 + 例子。 ### ❌ "请尽量准确" 这种废话 模型不会因为你 polite 就更努力。直接指令。 ### ❌ 给"超过 100k token" 的 context 大 context 时模型有"middle lost in the haystack" 效应——中间的信息 被忽略。能 chunk + RAG 的就别一次塞。 ## 调试 / 评估 写个 eval set（30-100 个 ground truth 例子）： ```python test_cases = [ {'ticket': '...', 'expected': 'billing'}, {'ticket': '...', 'expected': 'bug'}, ... ] correct = 0 for tc in test_cases: pred = classify(tc['ticket']) if pred['category'] == tc['expected']: correct += 1 print(f'accuracy: {correct/len(test_cases):.2%}') ``` 每改 prompt 跑一遍 eval。比"感觉好像准了" 客观。 用 LangSmith / Promptfoo / weave 等工具系统化跑 A/B prompt 对比。 ## 模型选择 不同任务用不同模型： | 任务 | 推荐 | |---|---| | 简单分类 / 提取 | gpt-4o-mini / claude-haiku（便宜 + 快） | | 推理 / 代码 | gpt-4o / claude-sonnet-4-5 | | 极致难推理 | o1 / claude opus-4-7 / DeepSeek-R1 | | 本地隐私 | qwen2.5:14b / llama3.1:8b（Ollama） | 强行用大模型做简单分类 = 浪费钱。 ## 效果 工单分类项目最终： - 准确率 60% → 92% （+ verifier 后 96%） - 单 ticket 成本 $0.012 → $0.003（batch + 改 mini） - P95 延迟 5s → 1.2s（mini + structured output） - 总 month cost 从 $800 → $150 ## 踩过的坑 1. **改 prompt 不跑 eval**："感觉变好了"是偏见。每次改后跑测试集 验证。 2. **prompt 越改越长**：加 patch 修 case → 长 prompt → 模型 confused。 定期 refactor 简化。 3. **生产环境模型版本变**：OpenAI / Anthropic 偶尔 silent 升级模型， prompt 行为变化。pin model name 包括 date suffix (`gpt-4o-2024-08-06`) + 监控 metrics。 4. **temperature**：分类 / 提取类 task 设 0 (deterministic)。 创意写作设 0.7-1.0。 5. **隐私数据**：不要把客户 PII 直接发 OpenAI。先 mask 邮箱 / 手机 / 信用卡再调 API。

## 起因 我们的 RAG 系统每天调 OpenAI `text-embedding-3-small` 几十万次，账单 $300+/月。而且内部知识库不能出公网。bge-m3 是智源开源的多语言 embedding 模型，质量在 MTEB 中文榜上经常排前 3，比 OpenAI 3-small 还好。 本地跑一张 4090 就够。 ## 解决方案：FastAPI 包一层 OpenAI 兼容接口 让现有用 OpenAI Python SDK 的代码改一行 base_url 就能切。 ### 装 ```bash uv add fastapi 'uvicorn[standard]' sentence-transformers # 第一次启动会下载 bge-m3 模型 (~1.2 GB) ``` ### service.py ```python import logging from contextlib import asynccontextmanager from typing import List import torch from FlagEmbedding import BGEM3FlagModel from fastapi import FastAPI, HTTPException from pydantic import BaseModel log = logging.getLogger('embedding') class EmbeddingRequest(BaseModel): input: str | List[str] model: str = 'bge-m3' encoding_format: str | None = 'float' class EmbeddingData(BaseModel): object: str = 'embedding' embedding: List[float] index: int class EmbeddingResponse(BaseModel): object: str = 'list' data: List[EmbeddingData] model: str usage: dict model: BGEM3FlagModel | None = None @asynccontextmanager async def lifespan(app: FastAPI): global model log.info('loading bge-m3 ...') model = BGEM3FlagModel('BAAI/bge-m3', use_fp16=torch.cuda.is_available()) log.info('ready') yield app = FastAPI(lifespan=lifespan) @app.post('/v1/embeddings', response_model=EmbeddingResponse) def embed(req: EmbeddingRequest): if model is None: raise HTTPException(503, 'model loading') texts = [req.input] if isinstance(req.input, str) else req.input if not texts: raise HTTPException(400, 'empty input') if len(texts) > 256: raise HTTPException(400, 'batch too large') out = model.encode(texts, batch_size=32, max_length=8192, return_dense=True)['dense_vecs'] data = [EmbeddingData(embedding=vec.tolist(), index=i) for i, vec in enumerate(out)] total_tokens = sum(len(t) for t in texts) // 4 # 粗估 return EmbeddingResponse( data=data, model='bge-m3', usage={'prompt_tokens': total_tokens, 'total_tokens': total_tokens}, ) @app.get('/health') def health(): return {'ok': model is not None} ``` ### 起服务 ```bash uv run uvicorn service:app --host 0.0.0.0 --port 8001 # 或生产： uv run gunicorn -k uvicorn.workers.UvicornWorker \ -w 1 -b 0.0.0.0:8001 service:app ``` 注意 `-w 1`：embedding 是 GPU 密集，多 worker 抢一张卡反而慢。 水平扩容用多机或者把模型放多张卡。 ### 客户端：OpenAI SDK 直接用 ```python from openai import OpenAI client = OpenAI( base_url='http://localhost:8001/v1', api_key='local', # 不校验，随便填 ) resp = client.embeddings.create( input=['你好世界', 'Hello world', '今日天气不错'], model='bge-m3', ) for i, d in enumerate(resp.data): print(f'{i}: dim={len(d.embedding)} first 5={d.embedding[:5]}') ``` ## 性能 我的 RTX 4090 上： - batch=32 文本（每条 ~200 字）：~80ms（≈ 400 texts/s 持续） - batch=1：~20ms P95 延迟 < 100ms 在企业 RAG 场景完全够。OpenAI 3-small 平均 200-500ms （网络），自托管反而更快。 ## 效果 - 月度 embedding 账单 $300 → $0 - P95 延迟 320ms → 80ms（无公网往返） - 内部敏感文档不出公网 - bge-m3 中文 MTEB 评测比 text-embedding-3-small 高 5-8 个点 ## 踩过的坑 1. **第一次冷启动慢**：bge-m3 模型 1.2 GB 从 HuggingFace 拉。在国内 设 `HF_ENDPOINT=https://hf-mirror.com`。 2. **OpenAI SDK 严格校验维度**：bge-m3 输出 1024 维，OpenAI 3-small 是 1536。如果客户端代码硬编码 1536 校验，要么改客户端要么改 model 多输出 padding。 3. **`use_fp16=True` 在某些老 GPU 数值溢出**：T4 / 老 CUDA 上偶尔 NaN。 降到 `use_fp16=False` 慢 30% 但稳。 4. **batch 太大显存爆**：bge-m3 max_length=8192，batch=64 × 8192 token 可能 OOM。生产 `batch_size=32` + `max_length=512`（多数 chunk 这个 长度够用）。 5. **multi-process 共享 GPU 模型不共享内存**：每个 worker 各自 load 一份 1.2GB。`-w 1` 单 worker 是正确选择；要并发用 async + 内部 batch 合并请求。

## 起因 要做一个"自然语言查数据库"的功能。用户问"上周日北京下单的用户有几个？" → LLM 生成 SQL → 后端执行 → 返结果。 最原始做法是让 LLM 生成 SQL 字符串然后 regex 提取。痛点： - 模型有时输出 ```sql ... ``` markdown 包裹 - 有时多输出一段"分析这个 SQL..." 散文 - 有时 SQL 语法错（缺逗号、wrong table） - parse 失败要 try/except 重试 `function calling`（OpenAI 起的名，Anthropic 叫 tool use）让模型 **直接结构化输出"调用什么函数 + 什么参数"**，零解析。 ## 解决方案：tool calling ### 定义工具 ```python from openai import OpenAI client = OpenAI() tools = [ { 'type': 'function', 'function': { 'name': 'run_sql', 'description': '在分析数据库上执行 SELECT SQL，返回最多 50 行。' '只允许 SELECT；DELETE/UPDATE/INSERT 会被拒绝。', 'parameters': { 'type': 'object', 'properties': { 'sql': { 'type': 'string', 'description': 'PostgreSQL 标准 SQL，必须以 SELECT 开头', }, 'explanation': { 'type': 'string', 'description': '一句话解释这个 SQL 在做什么', }, }, 'required': ['sql', 'explanation'], }, }, }, { 'type': 'function', 'function': { 'name': 'list_tables', 'description': '列出可用表名', 'parameters': {'type': 'object', 'properties': {}, 'required': []}, }, }, ] system_prompt = """ 你是一个数据分析助手。回答用户问题前，可能需要： 1. 用 list_tables 看有哪些表 2. 用 run_sql 查数据 数据 schema： - users(id, email, country, created_at, plan) - orders(id, user_id, amount, city, ordered_at, status) - products(id, name, category, price) 回答用户用中文。 """ def chat(messages): return client.chat.completions.create( model='gpt-4o', messages=messages, tools=tools, ) ``` ### 调用循环 ```python def run(user_question: str): messages = [ {'role': 'system', 'content': system_prompt}, {'role': 'user', 'content': user_question}, ] while True: resp = chat(messages) msg = resp.choices[0].message # 模型决定调函数 if msg.tool_calls: messages.append(msg) # assistant turn for tc in msg.tool_calls: result = dispatch(tc.function.name, json.loads(tc.function.arguments)) messages.append({ 'role': 'tool', 'tool_call_id': tc.id, 'content': json.dumps(result), }) # 继续下一轮，让模型看 tool 结果再决定 continue # 模型给最终答案 return msg.content ``` ### 实现 dispatch（真的执行 SQL） ```python import psycopg def dispatch(name, args): if name == 'list_tables': return list_tables() if name == 'run_sql': return run_sql(args['sql']) return {'error': f'unknown function: {name}'} def list_tables(): with psycopg.connect(DB_URL) as conn: cur = conn.execute(""" SELECT table_name, obj_description(...) FROM information_schema.tables WHERE table_schema='public' """) return [{'table': r[0], 'desc': r[1]} for r in cur.fetchall()] def run_sql(sql: str): if not sql.strip().upper().startswith('SELECT'): return {'error': 'only SELECT allowed'} try: with psycopg.connect(DB_URL, autocommit=False) as conn: conn.execute('SET statement_timeout=10000') # 10s 限时 cur = conn.execute(sql) rows = [dict(zip([c[0] for c in cur.description], r)) for r in cur.fetchmany(50)] return {'rows': rows, 'count': len(rows)} except Exception as e: return {'error': str(e)} ``` ### 跑一下 ```python print(run('上周日北京下单的用户有几个？')) # 输出： # 上周日（2024-05-19）在北京下单的用户共 47 个。 ``` 模型自动： 1. 调 `list_tables()` 看有哪些 2. 调 `run_sql('SELECT COUNT(DISTINCT user_id) FROM orders WHERE city=...')` 3. 拿到结果后用自然语言回答 整套流程**无字符串解析**——arguments 已经是 typed JSON。 ## 几个重要细节 ### 1. 多 tool 同时调 ```python if msg.tool_calls: # 可能一次调 2-3 个函数 for tc in msg.tool_calls: ... ``` 模型可能并行调 list_tables + run_sql。要 loop 处理所有。 ### 2. 防恶意 SQL 工具签名再严格也挡不住"DROP TABLE users; --" 写在 SQL 字符串里。 在 dispatch 层做实际验证： - 只允许 SELECT 开头 - 用只读 DB 用户（无 DDL / DML 权限） - `SET statement_timeout=N` 限时长 - ACL 限 schema / table 访问 - 用 SQLAlchemy `text(sql)` + 参数化（更难做） 或者更激进：sandbox 跑（DuckDB on read-only data copy）。 ### 3. 限制循环次数 模型可能死循环调工具。限步数： ```python for step in range(10): resp = chat(messages) if not msg.tool_calls: return msg.content # ... raise RuntimeError('exceeded 10 tool-use steps') ``` ### 4. parallel tool call OpenAI 默认开 parallel；要禁用： ```python resp = client.chat.completions.create(..., parallel_tool_calls=False) ``` 复杂任务有 dependency 时禁用更稳。 ### 5. tool_choice 强制 ```python client.chat.completions.create( ..., tool_choice={'type': 'function', 'function': {'name': 'run_sql'}}, ) ``` 强制本轮调某个函数（不让模型 freestyle 直接答）。 ## Anthropic / Gemini / Ollama 也支持 API 风格略不同但概念一致。 ```python # Anthropic import anthropic client = anthropic.Anthropic() resp = client.messages.create( model='claude-sonnet-4-5', max_tokens=1024, tools=[{'name': 'run_sql', 'description': '...', 'input_schema': {...}}], messages=[{'role': 'user', 'content': '...'}], ) # stop_reason='tool_use' 时遍历 content blocks ``` ```python # Ollama (qwen2.5 等支持 function calling) import ollama resp = ollama.chat( model='qwen2.5:7b', messages=[...], tools=[{'type': 'function', 'function': {...}}], ) ``` 跨家 LLM 工具调用接口已经形成事实标准（OpenAI 格式被大部分模仿）。 ## 实际应用场景 1. **数据库 query agent**（上面例子） 2. **代码 review bot**：tool 是 read_file / list_files / run_tests 3. **客服 agent**：lookup_order / refund / escalate 4. **DevOps agent**：check_deployment / rollback / fetch_logs 5. **RAG with citations**：search_docs / fetch_doc tool 任何"模型需要查外部信息再回答" 都适合。 ## 与 LangChain / LlamaIndex 的关系 LangChain `create_react_agent` / `create_tool_calling_agent` 是上面 循环的封装： ```python from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_openai import ChatOpenAI from langchain.tools import tool @tool def run_sql(sql: str) -> str: """Execute SELECT SQL, return rows.""" return run_sql_impl(sql) llm = ChatOpenAI(model='gpt-4o') agent = create_tool_calling_agent(llm, [run_sql], prompt) executor = AgentExecutor(agent=agent, tools=[run_sql], max_iterations=10) executor.invoke({'input': '上周日北京...'}) ``` 封装方便但藏了细节。简单场景手写循环更可控；复杂 agent 用 LangChain 省事。 ## 效果 我们的 SQL agent 上线后： - 业务团队不再用 Metabase 拖拽，直接问中文 - "为什么这个客户流失了" 类自由查询能 90% 准确给出 SQL + 结果 - function calling 解析失败率：0%（结构化 output） - vs 之前用 regex 提取 SQL 字符串：~12% 失败要重试 ## 踩过的坑 1. **参数 schema 错**：模型按 schema 生成参数，schema 不严会乱传。 `required` / `enum` / `type` 都明确写。 2. **大 tool 数 → 模型 confused**：超过 ~10 个 tool 后模型选错率 上升。分层：top-level "router" → 选 sub-agent → sub-agent 有 3-5 个 tool。 3. **tool 实现 crash 抛异常**：返 `{'error': str(e)}` 让模型看见， 模型会 retry / 换策略。直接 raise 就只能上层 catch。 4. **token 成本**：tool description + schema 占 system prompt 不少 token。每次请求都付。优化：精简 description / 用 short tool name。 5. **流式（streaming）+ tool**：streaming response 中 tool_call chunks 是分段的，要 buffer 后再 parse。复杂场景非流式更稳。

直接用 HuggingFace transformers 跑 LLM 推理性能很差： batch 1 时 GPU 利用率 30-50%，多并发请求时显存碎片化 OOM。 vLLM 是伯克利出的高性能 LLM 推理引擎，核心技术是 **PagedAttention** （像 OS 分页一样管理 KV cache），加上 continuous batching， 比 transformers 直接推理快 5-24 倍。 ## 安装 ```bash uv add vllm # 需要 CUDA 11.8+ 或 12.x，PyTorch 2.x ``` ## 命令行起服务 ```bash uv run vllm serve Qwen/Qwen2.5-7B-Instruct \ --tensor-parallel-size 1 \ --max-model-len 8192 \ --gpu-memory-utilization 0.85 \ --port 8000 ``` 第一次启动会从 HuggingFace 下载模型（~15GB）。 启动后默认 OpenAI 兼容 API。 ## 调用 ```bash curl http://localhost:8000/v1/chat/completions \ -H 'Content-Type: application/json' \ -d '{ "model": "Qwen/Qwen2.5-7B-Instruct", "messages": [{"role": "user", "content": "你好"}], "max_tokens": 200 }' ``` 或 Python： ```python from openai import OpenAI client = OpenAI(base_url='http://localhost:8000/v1', api_key='dummy') resp = client.chat.completions.create( model='Qwen/Qwen2.5-7B-Instruct', messages=[{'role': 'user', 'content': '你好'}], max_tokens=200, ) print(resp.choices[0].message.content) ``` ## 关键参数 - `--max-model-len`：上下文最大长度（影响 KV cache 大小） - `--gpu-memory-utilization`：用多少显存（0-1，默认 0.9） - `--tensor-parallel-size`：多 GPU 时拆 tensor 并行（4 卡设 4） - `--quantization awq` / `gptq` / `fp8`：量化加速 - `--enable-prefix-caching`：相同前缀的请求复用 KV cache（系统 prompt 共享场景大幅加速） ## Python 直接调用（不走 HTTP） ```python from vllm import LLM, SamplingParams llm = LLM(model='Qwen/Qwen2.5-7B-Instruct', gpu_memory_utilization=0.85) prompts = [ '介绍一下 RAG', '解释 PagedAttention', '写一个 Python 二分查找', ] params = SamplingParams(temperature=0.3, max_tokens=300) outputs = llm.generate(prompts, params) for out in outputs: print('---') print(out.outputs[0].text) ``` vLLM 自动 batch 这 3 个 prompt 一起跑，单次 forward 处理多个序列。 ## continuous batching 的含义 传统推理： ``` batch 1: [seq A 100 tokens, seq B 80 tokens, seq C 60 tokens] 等三个序列都完成才能下一个 batch ``` continuous batching： ``` 任意时刻一个请求结束就立刻让出位置给新请求 GPU 持续吃满，无 idle ``` 这是 vLLM 高吞吐的核心，比"动态 batch"更激进。 ## benchmark：vs 直接 transformers ```bash # 100 个并发请求，每个生成 200 token # vLLM ab -n 100 -c 16 -p body.json -T application/json \ http://localhost:8000/v1/chat/completions # 通常：3000-8000 tokens/s 吞吐 # transformers + 简单 FastAPI 包装 # 通常：300-800 tokens/s ``` 10x 量级的吞吐差距。 ## 多卡：tensor parallelism 70B 模型单卡装不下，4 张 A100 拆开： ```bash uv run vllm serve meta-llama/Llama-3.1-70B-Instruct \ --tensor-parallel-size 4 \ --max-model-len 8192 ``` vLLM 自动用 NCCL 在 4 卡间分配 attention head / FFN 矩阵。 ## 量化：让更大模型跑在更小显卡 ```bash # AWQ 4-bit uv run vllm serve TheBloke/Llama-3.1-70B-AWQ \ --quantization awq # 4 bit 量化的 70B 大约 40GB 显存（不量化要 140GB） ``` ```bash # FP8 (需要 H100) uv run vllm serve meta-llama/Llama-3.1-70B-Instruct \ --quantization fp8 ``` ## 长上下文 ```bash # 32k 上下文 uv run vllm serve Qwen/Qwen2.5-7B-Instruct \ --max-model-len 32768 ``` 但 KV cache 占显存 = batch_size × max_seq_len × 每层 KV size。 32k context + 100 batch ≈ 显存吃 50%+，要 trade off。 ## 与 Hugging Face 模型生态 vLLM 支持的 model：Llama / Mistral / Qwen / Mixtral / Gemma / Yi / DeepSeek / Phi / Baichuan / ChatGLM 等几乎全部主流开源 LLM。 官方维护清单看 vLLM docs。 ## 与 sglang / lmdeploy 对比 | 引擎 | 优势 | 劣势 | |---|---|---| | vLLM | 生态最大、模型最多 | 长上下文性能一般 | | TGI (HF) | HF 官方、生产稳 | 吞吐略低于 vLLM | | sglang | 结构化生成（JSON / regex）极快 | 模型支持稍少 | | lmdeploy | 国内（商汤）、TurboMind 后端快 | 文档不全 | 通用选 vLLM；要求 JSON 严格输出选 sglang。 ## prefix caching：相同系统 prompt 复用 ```bash uv run vllm serve qwen2.5:7b --enable-prefix-caching ``` 所有请求都用 "You are a helpful assistant..." 起头的话， prefix 这部分的 KV cache 只算一次，10k token 系统 prompt 几乎免费。 ## 生产部署清单 1. 用 systemd 起 vLLM service 2. 前面套 nginx 反代（限流 + auth） 3. Prometheus 抓 vLLM 内置的 `/metrics` 4. health check：`/health` 5. 多模型用多个 vLLM 进程，每个绑不同 GPU ## 踩过的坑 - 启动时 "out of memory"：`--gpu-memory-utilization` 调小， 或减 `--max-model-len`。 - 模型权重 download 慢：用 HuggingFace mirror 或预先下载， `HF_HUB_OFFLINE=1` 让 vLLM 不再尝试下载。 - TP > 1 时 NCCL 卡死：检查机器内 GPU 互联（PCIe / NVLink）； 设 `NCCL_P2P_DISABLE=1` 排查。 - vLLM 0.5+ 跟 PyTorch 2.4+ 紧耦合，旧 PyTorch 装不上。`uv` 自动解析 依赖一般没问题，手动 pip 时容易翻车。