LangChain -AI编程框架（更新中...
ps：LLM AI编程
25年底 面试如果你能讲出来agent 是一个加分项 技术面会觉得你很不错。
26年目前吧 后端面试来说 翻阅网上大大小小的面经 agent已经逐渐变成了一个必备项。
AI应用工程师=传统后端接入LLM
是什么
是连接ai模型和实际ai应用的一个桥梁
langchain
让大模型不仅是一个百科全书，而是成为一个能行动，能思考，能进行业务流程的合作伙伴（agent）
认识模型
模型就是从数据中找到规律 学习规律的数学函数或程序。
认识大语言模型LLM
大语言模型（LLM）就是以海量文字为数据，专门学习人类语言规律、知识、逻辑、语法、表达习惯的超级人工智能模型。
拆解理解
训练数据喂给它亿万级的书本、文章、对话、网页、百科等所有人类文字内容。
核心学习的规律文字怎么组词、句子怎么通顺、上下文逻辑是什么、知识是什么、人类怎么思考和表达。
本质一个超大号的语言数学函数 + 程序，靠学到的语言规律，自动理解文字、续写文字、回答问题、创作内容、逻辑推理。
简单类比
普通小模型：只会做单一事（比如只识别图片、只算数据）
大语言模型：精通人类语言，能聊天、答题、写文案、翻译、总结、思考、写代码，通用能力极强。
LLM的能力
1.语言大师
2.知识巨人
3.逻辑与代码
4.多模态先知
“多模态AI”指的是能同时处理并理解多种类型数据（如文字+图片+声音）的模型。
提示词编写
1.costar结构化框架
可以把CoSTAR理解为一个“有效指令模板”，它通过六个维度将模糊的任务拆解成AI易于理解的清晰指令。
要素 含义 核心作用 (Why?) 实操关键 (How?) C (Context) 上下文/背景 为AI提供任务的前提信息和场景，避免它做出泛泛的、脱离背景的回答。 “你现在是一名资深的科技记者。” O (Objective) 目标/任务 明确告诉AI你需要它具体达成什么目的，让AI的“大脑”集中火力，使命必达。 “你需要为我们的新产品写一篇宣传稿，目的是吸引投资。” S (Style) 风格 指定AI回应的文风、措辞和格调，让输出内容更专业、更具感染力。 “模仿著名科技分析师Ben Thompson的写作风格。” T (Tone) 语气 设定AI回复所应传达的情感基调，这直接影响信息被接收时的感受。 “语气需要专业、客观，同时富有前瞻性。” A (Audience) 受众 明确你的内容是为谁准备的，让AI可以调整信息的深度、广度和专业术语的使用频率。 “你的报告要面向公司董事会成员。” R (Response) 输出格式 指定你想要的输出格式，比如JSON、Markdown或项目符号列表，让结果立即可用。 “请将最终信息以JSON格式输出。” 相比之下，其他框架也会从不同角度来定义有效指令。例如，有框架将CoSTAR的结构定义为背景、目标、策略、战术、行动和结果；还有框架提供了类似但步骤有所不同的简化模型（如STAR）。它们本质上都是希望将复杂、模糊的人类语言需求，转化为AI可以精准理解的“结构化任务书”。
假设你在经营一家公司，想让AI帮你写一篇社交媒体帖子来推广新产品。
普通指令（效果差）：
“帮我写篇帖子宣传我们公司的新吹风机，叫'极风'。”CoSTAR指令（效果好）：
text
Context (上下文) 我是一家智能家电创业公司的市场专员，我们公司的新产品是“极风”吹风机，主打“3分钟快干”和“智能温控护发”。 Objective (目标) 为这款新产品撰写一篇抖音/小红书风格的宣传帖。核心目标是吸引20-30岁的年轻女性用户，并引导她们访问产品预售页面。 Style (风格) 参考“完美日记”或“蕉内”等消费品牌的年轻化、趣味性短文案风格。 Tone (语气) 充满活力，带一点幽默感和生活方式气息，比如“懒人福音”、“吹头发不再像举铁”。 Audience (受众) 20-30岁的都市年轻女性，她们注重效率和生活品质，可能遇到过吹头发慢、损伤发质的问题。 Response (输出格式) 请直接输出完整的文案，文案分为三部分： 1. 一个吸引眼球的标题。 2. 正文内容（不超过150字），用短句和表情符号表达。 3. 结尾用一句行动号召语。将模糊的日常语言，变成结构清晰、指令明确的“需求文档”，大部分情况下能让LLM给出更高质量、更贴合需求的答案。
比如我们来验证一下 我用我的deepseek开启两个窗口
这是没有提示词的
这是加上提示词的
2.少样本提示
一样道理我们测试另一种
有提示后
3.思维链提示
思维链提示是一种用于引导大型语言模型进行复杂推理的技术。
你可以把它理解为：要求模型在给出最终答案之前，先把解题的中间思考过程一步一步写出来。
核心思想
不只是得到答案：让模型展示出它是如何推导出答案的，让推理过程清晰可见。
把复杂问题拆分：一个大问题被分解成多个小的、易处理的步骤，模型的每一步都基于上一步的结果。
提高准确性：鼓励模型进行更系统、更符合逻辑的思考，减少直觉性错误。
举个例子
假设问模型：“一家商店买进苹果，进价每个5元，售价每个8元。如果卖出15个苹果，商店获利多少元？”
没有使用思维链提示（模型可能直接猜）：
答：45元。（可能是错误的）使用思维链提示（模型会这样思考并回答）：
每个苹果的利润 = 售价 - 进价 = 8元 - 5元 = 3元。
卖出15个苹果的总利润 = 每个利润 × 数量 = 3元 × 15 = 45元。
因此，商店获利 45元。
虽然结果是相同的，但通过思维链提示，我们确保了逻辑正确，并且可以检查中间步骤是否有误。
总结表格
方面 描述 定义 要求模型展示出逐步推理过程，然后给出最终答案 目的 提高复杂推理任务的准确性、可解释性 实现方式 在提示中加入“逐步思考”等指令 适用场景 数学题、逻辑推理、多步问题、决策分析 核心价值 透明、可验证、减少错误
4.自动推导和零样本链式思考
零样本链式思考 (Zero-shot CoT)
这是最常用、也最便捷的一种思维链技术。
“零样本”的含义：指你不需要在提示中给模型提供任何例子。你不需要先演示“问题A，步骤1，步骤2，答案A；问题B，步骤1，步骤2，答案B”，而是直接提出你的问题。
“链式思考”的实现：你只需在问题后面加上一句特定的魔法短语，来“激活”模型的逐步推理能力。最经典的魔法短语就是：“让我们一步步思考” (Let's think step by step)。
工作原理：模型看到这句指令后，会被引导先去生成一个包含中间推理步骤的文本，然后再基于这个推理过程生成最终答案。
例子：
用户提问：
一个水池有一个进水管和一个出水管。进水管单独注满水池需要4小时，出水管单独排空水池需要6小时。如果两个水管同时打开，多少小时能注满水池？让我们一步步思考。模型就会先输出推理过程，再给出答案。
优点：极其简单，无需准备示例，适用于处理从未见过的问题。
2. 自动推导 (Automatic Reasoning / Auto-CoT)
这个概念是为了解决“零样本CoT”的一个小缺点而提出的。这个缺点是：零样本CoT生成的推理步骤质量有时不稳定，可能会出错。
核心思想：让模型自己给自己生成例子，而不是由人来编写例子。
工作原理：
问题聚类：将你手头的一堆问题（例如10个数学题），根据它们的语义或类型进行分组。
生成推理链：对每组中的一些代表性题目，使用零样本CoT（比如用“让我们一步步思考”指令）让模型生成每个问题的推理过程和答案。注意，这些生成的例子并不保证100%正确。
构建范例：将上一步生成的“问题-推理链-答案”作为“少样本”范例。
最终推理：当你有一个新问题时，你就在提示中放入这些由模型自己生成的范例，让模型参考它们来进行推理。
为什么叫“自动”：因为整个过程——从选择哪些问题作为例子，到生成这些例子的推理路径——都是模型自动完成的，不需要人工编写任何高质量的例子。
优点：比人工编写少样本提示更省力；比零样本CoT的推理稳定性更强，因为它给模型提供了参考范例。
核心区别与联系
为了方便你对比，这里整理了一个表格：
特性
零样本链式思考 (Zero-shot CoT)
自动推导 (Auto-CoT)
是否需要示例
不需要。直接提问 + 魔法短语。
需要。但示例是模型自己生成的。
示例来源
无
由模型通过零样本CoT自动生成
提示复杂程度
极低
较高（需要先让模型为自己生成一批示例）
推理稳定性
一般。复杂问题可能出错。
更高。有示例做参考，更稳定。
适用场景
快速上手、简单到中等复杂的问题
对准确性和稳定性要求更高的批量任务
经典魔法短语
“Let's think step by step.”
无固定短语，核心是使用检索/聚类 + 生成示例的流程
总结
零样本CoT就像是直接问模型：“你告诉我答案，但要把你的思考过程写出来。” 是最简单、最常用的技巧。
自动推导（Auto-CoT）就像是让模型先看一些“自己以前做过的类似题目和解题过程”作为参考，然后再去做新题。 是更高级、更稳定的技术。
区别
总结：一句话分清
思维链提示 = “把思考过程写出来” （这是一个指令或目标）。
零样本链式思考 = 直接问，让它当场想 （不教它，让它自己推理）。
自动推导 = 先让它自己学几个例子，再用例子去教它做新题 （自动构建示范，提高质量）。
所以，“‘自动推导’和‘零样本链式思考’与思维链提示区别在哪？”
区别在于层级：思维链提示是父集（总概念）。
自动推导和零样本链式思考是思维链提示这个总概念下的两个具体子方法，它们在是否提供示例以及示例从何而来上有显著区别。
5.自我评判与迭代
自我评判：模型自己评估自己生成的推理过程或最终答案，判断是否存在错误、遗漏或逻辑不一致。通常包括正确性打分、错误定位、风险识别等。
迭代：基于自我评判的结果，模型对原有推理进行修正、补充或重写，然后再次进行评判，如此循环，直到满足某个停止条件（如达到最大迭代次数、评判通过、答案不再变化等）。
这两者通常结合在一起使用，形成 “生成 → 评判 → 反馈 → 再生成” 的闭环。
工作流程
1. 初始生成
↓
2. 自我评判（发现问题？）
↓ 是
3. 基于反馈进行迭代修正
↓
4. 再次评判（问题解决？）
↓ 是
5. 输出最终答案
（如果否，则返回步骤3，继续迭代，直到达到最大迭代次数）
实际例子
问题：一个长方形的长是宽的2倍，周长是36厘米。求面积。
仅使用思维链（一次生成）：
设宽为w，长为2w。周长=2(w+2w)=6w=36 → w=6，长=12。面积=6×12=72平方厘米。✓（恰好正确）
需要自我评判与迭代的例子（假设模型第一次犯了错）：
第一轮生成：
设宽为x，长为2x。周长= x + 2x = 36 → 3x=36 → x=12，长=24。面积=12×24=288。自我评判：
错误：周长公式用错了，应该是2×(长+宽)，而不是长+宽。因此第一步计算错误。
评分：3/10。迭代修正：
正确解法：设宽为w，长为2w。周长=2(w+2w)=6w=36 → w=6，长=12。面积=6×12=72。再次自我评判：
正确，评分10/10。输出最终答案72。
LLM接入方式
当我们自己构建AI应用时候，无法直接使用其客户端，就需要通过代码的方式
接入原生LLM的能力。
API远程调用
先下载Apifox和注册账号
我用的豆包 deepseek都可以
豆包（火山方舟 Ark）API 接入 总体流程大纲
一、前期准备：开通服务与获取凭证
注册与实名认证
注册火山引擎账号，完成个人实名认证（必须，否则无法使用 API 服务）。
进入「火山方舟 Ark」控制台。
创建并获取 API Key
进入左侧菜单「API Key 管理」，点击「创建 API Key」。
记录生成的 API Key（格式：
ark-xxxx-xxxx-xxxx-xxxx），这是后续认证的核心凭证。授权并开通模型服务
进入「在线推理」-「预置推理接入点」。
找到你想使用的模型（如
doubao-seed-1-8/doubao-lite-4k），点击「授权」，同意活动规则，获取免费额度。获取模型 / 接入点 ID
授权后，在「快速接入测试」的示例代码中，找到模型对应的 ID（如
doubao-seed-1-8-251228）。也可以通过「自定义推理接入点」创建独立接入点，获取格式为
ep-xxxx-xxxx-xxxx-xxxx的接入点 ID。
二、API 调试：在工具中配置请求（以你用的调试工具为例）
配置请求地址与方法
请求方法：
POST请求地址：
https://ark.cn-beijing.volces.com/api/v3/chat/completions（兼容 OpenAI 格式）或官方原生地址：
https://ark.cn-beijing.volces.com/api/v3/responses配置请求头（Headers）
表格
参数名
参数值
Content-Type
application/json
Authorization
Bearer API Key配置请求体（Body）
json
{ "model": "你的模型/接入点ID", "messages": [ { "role": "user", "content": "你好，介绍一下自己" } ] }发送请求并验证结果
我这种情况就调用成功啦
你们如果失败可以问问ai
有可能没额度 有可能没填对ID
开源模型本地部署
大模型本地部署，这种方式就是将开源的⼤型语⾔模型（如 Llama、ChatGLM、Qwen 等）部署在你自己的硬件环境（本地服务器或私有云）中。核⼼概念就是，将下载模型的⽂件（权重和配置文件），使⽤专门的推理框架在本地服务器或 GPU 上加载并运⾏模型，然后通过类似 API 的方式进行交互
典型流程是：
1. 获取模型：从 Hugging Face（国外）、魔搭社区（国内）等平台下载开源模型的权重。（平台参
考本篇章第四节）
2. 准备环境：配置具有⾜够显存（如 NVIDIA GPU）的服务器，安装必要的驱动和推理框架。
3. 选择推理框架：使⽤专为⽣产环境设计的框架来部署模型，
例如：
vLLM：特别注重⾼吞吐量的推理服务，性能极佳。
TGI：Hugging Face 推出的推理框架，功能全⾯。
Ollama：⾮常⽤⼾友好，可以⼀键拉取和运⾏模型，适合快速⼊⻔和本地开发。
LM Studio：提供图形化界⾯，让本地运⾏模型像使⽤软件⼀样简单。
4. 启动服务并调⽤：框架会启动⼀个本地 API 服务器（如 http://localhost:8000 ），你可以
像调⽤云端 API ⼀样向这个本地地址发送请求。
SDK和官方客户端库
这并⾮⼀种独⽴的接⼊⽅式，⽽是对第⼀种 API 接⼊的封装和简化。模型提供商通常会发布官⽅编程语⾔ SDK，为我们封装好了底层的 HTTP 请求细节，提供⼀个更符合编程习惯的、语⾔特定的函数库。典型流程（以 OpenAI Python SDK 为例）：安装库：pip install openai安装 OpenAI SDK 后，可以创建⼀个名为 example.py 的⽂件并将⽰例代码复制到其中：
from openai import OpenAI client = OpenAI(api_key="your-api-key") response = client.responses.create( model="gpt-5", input="介绍⼀下你⾃⼰。" ) print(response.output_text)相⽐直接构造 HTTP 请求，代码更简洁、更易读、更易维护。
LangChain也可以接入，甚至更多
如何选择
对于以上三种接⼊⽅式，我们该如何选择？看数据敏感性：如果数据极其敏感，必须留在内部，本地部署是唯⼀选择。看技术实⼒和资源：如果团队没有强⼤的 MLops（机器学习运维）能⼒，也没有预算购买和维护 GPU 服务器，云端 API 是更实际的选择。看成本和规模：如果应⽤规模很⼤，⻓期来看，本地部署的固定成本可能低于持续的 API 调⽤费⽤。反之，⼩规模应⽤ API 更划算。看定制需求：如果只是使⽤模型的通⽤能⼒，云端 API ⾜够。如果需要⽤⾃⼰的数据微调模型，则需要选择⽀持微调的 API 或直接本地部署。
嵌入模型
什么是嵌入模型
⼤语⾔模型是⽣成式模型。它理解输⼊并⽣成新的⽂本（回答问题、写⽂章）。它内部实际上也使⽤嵌⼊技术来理解输⼊，但最终⽬标是“创造”。⽽嵌⼊模型（Embedding Model）是表⽰型模型。它的⽬标不是⽣成⽂本，⽽是为输⼊的⽂本创建 ⼀个最佳的、富含语义的数值表⽰（向量）。由于计算机天⽣擅⻓处理数字，但不理解⽂字、图⽚的含义。嵌入（Embedding）的核⼼思想就是将⼈类世界的符号（如单词、句⼦、产品、⽤⼾、图⽚）转换为计算机能够理解的数值形式（即向量本质上是⼀个数字列表），并且要求这种转换能够保留原始符号的语义和关系。
我们可以把它想象成⼀个翻译过程，把⼈类语⾔“翻译”成计算机的“数学语⾔”。
结论：既然是“数学语⾔”，那么我们可以⽤数学的⽅式来⽐较向量，从⽽达到【度量语义】的⽬的！
它的核心作用就一个：把文字转换成计算机能计算的“坐标”，从而判断语义的相似度。
基于这个能力，它有三大关键用途：
搜索引擎：把你的搜索词转成向量，去和网页库里的向量比对，找出意思最像的结果。现在的“语义搜索”能搜出不含关键词但意思相近的内容，靠的就是它。
知识库问答：这是最典型的应用。你先用嵌入模型把几千页文档都转成向量存起来；当你提问时，模型先把你的问题转成向量，快速找出相关的几段文字，再把这几段文字扔给大模型（如GPT）生成答案。没有嵌入模型，大模型面对海量文档会直接“看不过来”。
推荐与聚类：
推荐：算出你看过的文章向量，然后去找其他向量相近的文章推给你。 聚类：把成千上万条客服反馈自动分成“物流问题”、“质量吐槽”等几堆，无需人工打标签。一句话总结：大模型负责“说话”，嵌入模型负责“找资料”。没有嵌入模型，大模型处理超出记忆范围的业务（比如分析自家公司所有合同）时，就会束手无策。
应用场景
一、与大语言模型配合（这是目前最火的应用）
这是构建基于私有数据的AI应用（所谓RAG，检索增强生成）的基础。
企业知识库问答
场景：公司想把内部几千页的报销制度、技术文档、会议纪要做成一个问答机器人。
嵌入模型作用：先把所有文档切成小段，每段转成向量存储。当员工问“出差住宿标准是多少？”时，嵌入模型把问题也转成向量，迅速找到最相关的那几段文档规则。
结果：把这些相关段落交给大模型，大模型基于此生成准确的答案。
长文档总结
场景：一份100页的财报，直接扔给大模型会超出它的上下文长度限制。
嵌入模型作用：将文档分割成多个片段进行向量化，然后筛选出包含最关键数据的几个片段（哪怕这些片段分散在文档的不同位置）。
结果：只把精华片段送入大模型，让大模型进行总结，既省Token又保证了总结的完整性。
二、独立应用（不依赖大模型）
很多传统AI任务里，嵌入模型本身就是核心。
语义搜索引擎
场景：电商搜索“舒适的衣服”，传统搜索引擎只能找包含“舒适”或“衣服”字眼的商品，会漏掉“纯棉家居服”、“柔软睡衣”等。
作用：嵌入模型能理解“舒适”和“柔软、纯棉、家居”在语义上非常接近，从而把真正相关的商品搜出来，哪怕商品描述里没有“舒适”这个词。
推荐系统
场景：用户看完了一篇关于“Python编程技巧”的文章，需要推荐下一篇文章。
作用：把用户读过的文章向量化，在库里寻找与之夹角最小的其他文章。它不依赖标签，而是真正读懂文章内容，推荐“JavaScript入门”或者“算法面试题”。
文本聚类
场景：某论坛运营收到了1000条用户反馈，想知道大家都在抱怨哪几类问题。
作用：将1000条反馈转为1000个向量，利用算法自动聚合。例如，500个向量聚成一堆（全是问登录问题的），300个聚成一堆（全是问退款到账时间的）。
异常检测
场景：检测电商平台的恶意刷单评论（垃圾文本）。
作用：正常的评论向量会在空间里聚在一起，而“好评返现”、“广告链接”等垃圾文本的向量会很奇怪，会远离正常群组，从而被轻松标记出来。
三、多模态场景（不仅限文本）
嵌入模型也可以用于处理图片、音频。
以图搜图：把图片也变成向量。用户上传一张自拍，系统找出服装店库里最像的几件衣服。
零样本分类：比如你想把新闻分为“体育”、“政治”、“娱乐”。不用标注数据训练，只需把“体育”、“政治”、“娱乐”这几个词也变成向量。新来一篇新闻，转成向量后，看它离哪个词最近就归哪类。
总结
应用领域 核心逻辑 典型例子 RAG/问答 找相似段落给大模型 ChatPDF、公司内部问答机器人 搜索 理解语义，而非匹配关键词 淘宝/谷歌语义搜索、论文检索 推荐 找相似内容的物品 知乎文章推荐、网易云音乐心动模式 聚类 把相似的自动归为一堆 新闻分类、用户评论整理 异常检测 找不相似的离群点 垃圾邮件检测、恶意评论识别
主流的模型
下面是几款在特定领域表现突出的模型，以及各自擅长的领域：
模型名称 核心优势 关键特点 适用场景 微软 Harrier 综合能力最强
在权威MTEB-v2榜单排名第一，比第二名高出4分以上。完全开源，支持100+种语言，32k上下文窗口。提供27B/0.6B/270M三种规格，兼顾性能与效率。 追求顶尖检索精度、处理多语言、需要自研部署的企业级RAG应用。 谷歌 Gemini Embedding 2 全能选手与跨语言专家
在跨语言检索测试中准确率高达99.7%。原生支持多模态。闭源API服务，输出维度3072。能直接处理文本、图像、视频、音频和PDF。 多语言知识库、欧美市场业务、需要处理多种格式文件（如图文混排PDF）的应用。 微软 e5 系列 (e5-small, base, large) 性价比之王与速度冠军
在精确产品匹配的测试中，Top-5准确率达到100%。- e5-small: 超轻量(118M参数)，速度快16ms，适合实时搜索。
- e5-base: 平衡之选。
- e5-large: 追求更高精度。国内强烈推荐。电商搜索、推荐系统、对响应速度要求极高的检索场景。 阿里巴巴 Qwen3-VL-2B 开源多模态冠军
在图文匹配任务中，性能超过了谷歌等闭源API。开源且轻量（20亿参数），模态差距小（0.25），意味着图文对齐能力极强。 构建中文或中英双语的“以图搜图”、图文混合检索的系统。 NVIDIA Llama-embed-nemotron-8B 精准度专家
在Top-1准确率上表现突出（62%），即最擅长把最相关的文档排在第一位。开源，基于Llama 3.1。需要较好的GPU硬件支持，推理速度相对较慢（约200ms）。 对结果排序质量要求极高、能接受一定延迟的金融、法律等专业领域问答。
维度决定了嵌入模型“看待世界”的精细程度。
我们可以用一个生动的比喻来理解：
低维度（比如2维或3维）：就像只能用“东西南北”和“上下”来描述一个物体的位置。能说清，但很粗糙。苹果和橘子可能离得很近，但很难区分“红富士苹果”和“嘎啦苹果”。
高维度（比如768维、1536维、3072维）：就像能用几百个坐标轴来描述。除了位置，还能描述颜色、形状、口感、产地、价格、营养……几乎每个微妙的特征都有一个专属的维度来衡量。
接入方式
嵌入模型的接入方式主要有三种，按从易到难排序如下：
使用云服务API（最推荐）
做法：直接调用OpenAI、谷歌、阿里等厂商的HTTP接口。
优点：无需硬件，几行代码即可接入，性能强大。
缺点：数据需上传云端，长期大规模调用成本较高。
适合：快速开发、项目初期、算力有限的团队。
本地部署运行
做法：下载开源模型（如BGE、E5），使用
transformers库或专用工具text-embeddings-inference在本地GPU运行。优点：数据安全（不出内网），无API调用费，延迟可控。
缺点：需要GPU服务器，维护较复杂。
适合：金融医疗等数据敏感场景、大规模高频调用。
通过向量数据库集成
做法：使用Milvus、Pinecone、Qdrant等数据库，它们内置嵌入模型接口，数据入库时自动向量化。
优点：存储、索引、检索一站式解决，开发效率最高。
缺点：受限于数据库支持的模型种类。
适合：搭建完整的RAG或相似性搜索系统。
快速决策：新手选 API；数据机密选本地；做搜索系统选向量数据库。
Vibe Coding
是什么
“Vibe Coding”（氛围编程）是2025年初由前OpenAI联合创始人Andrej Karpathy提出的一个软件开发新概念。它描述了一种全新的编程范式：开发者不再手动编写代码，而是通过自然语言向AI描述意图和需求，由AI（如Cursor、Claude Code等工具）生成、修改和运行代码，开发者则主要负责验收结果和把控方向。
核心理念：从"写代码"到"定感觉"
如果说传统编程是"亲手建房子"，Vibe Coding就像"做甲方"——你告诉AI你想要的感觉（Vibe），AI去画图纸、买材料、施工，你只需要验收成品是否符合预期。
这个模式的核心运行逻辑是一个闭环流程：描述意图 → AI生成 → 感觉校验(Vibe Check) → 反馈迭代 → 确认交付。
在这一过程中，开发者的角色发生了根本变化：
维度
传统编程
Vibe Coding
开发者角色
代码编写者、调试者、架构师
创意引导者、需求架构师、结果验收官
核心能力
语法掌握、算法理解、调试能力
需求拆解、意图表达、结果校验、产品思维
核心流程
需求→设计→编码→调试→测试 线性流程
意图→AI生成→校验→迭代 闭环循环
关注焦点
代码如何实现
结果是否符合预期
Vibe Coding的局限性
虽然Vibe Coding能极大地提升原型开发的效率，但它在当前阶段存在不容忽视的局限和风险：
代码质量与可维护性堪忧
由于是通过对话式迭代生成，代码往往缺乏清晰、统一的结构和设计模式。变量、函数命名可能不一致，导致代码难以阅读、理解、调试和扩展。一项学术研究将此现象称为"速度与质量的权衡悖论"（speed-quality trade-off paradox），即开发很快，但产出的代码通常"快但有缺陷"。剑桥大学和微软的研究也指出，Vibe Coding并未消除对专业知识的需求，而是将其转移到了上下文管理、代码快速评估和决策上。安全与可靠性风险
开发者（尤其是非技术背景的）如果"跳过"对代码的审查，选择直接"全部接受"，可能会让安全漏洞、不安全的依赖项或逻辑错误直接进入生产环境。AI模型有时会生成不存在的函数库或错误的配置（即"幻觉"问题），这在复杂项目中会破坏连续性并削弱信任。对于涉及敏感数据或关键业务的核心系统，这种"放手"的开发方式风险极高。并不适用于所有场景
Vibe Coding非常适合原型验证、个人项目、内部工具或"周末一次性项目"。但当项目逻辑变得复杂、需要跨越多个模块或考虑高并发、高可用等企业级要求时，AI的上下文限制和理解偏差会使其表现挣扎。行业实践表明，要让它生成"生产级"软件，必须结合人工的架构约束、设计指导和严格的测试流程。思特沃克（Thoughtworks）的实验发现，在纯粹的Vibe Coding模式下（不施加任何设计和测试规则），AI在后续需求变更和调试时遇到了显著困难。并非"零门槛"，可能导致能力退化
很多人误以为Vibe Coding让编程变得毫无门槛，任何人都能成为开发者。实际上，它只是转移了技能门槛，而非消除它。一个缺乏编程基础的人，在遇到AI生成的复杂bug或需要微调AI无法解决的细节问题时，会束手无策。这种依赖还可能造成开发者（尤其是新手）底层调试能力、系统架构设计能力的退化。
可以看到，Vibe Coding 不是⼀个替代品，⽽是⼀个强⼤的效率倍增器。它的正确定位是：糟糕的程序员 优秀的辅助工具
AI开发框架
1. 定义
AI开发框架是构建、编排和部署人工智能应用（尤其是大模型应用）的标准化工具集与最佳实践集合。它封装了底层复杂性，提供了模块化、可重用的组件。
2. 核心战略价值
它已成为AI应用开发的“新战略高地”，其价值体现在：
降本提效：屏蔽不同AI模型、数据库的底层API差异，预置通用解决方案，大幅减少重复工作。
化繁为简：将复杂的任务（如检索增强生成RAG、多智能体协作）拆解为标准化的可组装模块（如链、图、工具）。
确立架构：为开发“不确定性”的AI应用提供确定的架构模式（如状态机、工作流），使项目可设计、可维护。
区分层次：掌握框架是从“能调用API”走向“能设计生产级系统”的关键分水岭，也是判断AI生成代码质量的依据。
3. 主流框架速览表
类别 核心框架 一句话定位 底层训练 PyTorch 学术界研究和灵活原型开发的首选。 TensorFlow 工业级大规模部署和端侧推理的王者。 应用编排 LangChain 最流行的通用AI应用编排框架（“乐高积木”）。 LangGraph 专注打造高可靠、可精细控制的多智能体系统。 专用优化 LlamaIndex RAG应用（文档问答、知识库）的首选框架。 CrewAI 让多个AI角色（Agent）协作完成复杂任务。 零代码/低代码 Dify 企业级一站式的可视化AI应用构建平台。 Coze (扣子) 字节出品，个人快速创建聊天机器人的利器。 跨语言/生态 Spring AI 让Java/Spring开发者用熟悉的方式集成AI。 4. 核心设计原则
抽象与封装：隐藏复杂实现，提供统一、简洁的接口。
模块化与可组装性：组件像乐高积木，可按需灵活拼装成复杂工作流。
关注点分离：将数据加载、提示词管理、模型调用、输出解析、任务流程等不同职责分离。
5. 与Vibe Coding（AI编程）的关系
互补而非替代：AI代码生成工具负责快速产出“零件”（代码片段、函数）。AI开发框架则提供了“装配蓝图”和“流水线”，是专业开发者进行架构设计、系统权衡、质量控制的核心能力。学习框架，正是为了能判断AI生成的代码是否“合格”。
举例
再不理解的话 我给你们举一个形象的例子。
AI开发框架就像一个“智能餐厅的后厨总控系统”。
展开这个比喻：
你（开发者）是餐厅老板/主厨：你懂菜品（业务需求），但不需要亲自洗菜、切菜、调火候。
大模型（如ChatGPT）是“顶级厨师”：他厨艺高超（生成内容、推理能力强），但他只负责炒菜，不会自己洗菜、摆盘、叫外卖骑手。
AI开发框架就是“后厨总控系统”：它把整个后厨串起来：
食材采购 → 对应数据加载（从数据库、PDF里拿信息）
洗菜切菜 → 对应数据预处理（把杂乱信息整理好）
调料架 → 对应提示词模板（告诉厨师怎么做这道菜）
传菜窗口 → 对应输出解析（把厨师炒好的菜装盘）
多个厨师协作 → 对应多智能体（一个切菜、一个炒菜、一个摆盘）
外卖接口 → 对应工具调用（查天气、发邮件、调API）
没有这个系统时（裸调API）：
你要自己：买菜→洗菜→切菜→递菜→等厨师炒→自己装盘→自己端出去。
结果：做个番茄炒蛋还行，做满汉全席（复杂应用）就会手忙脚乱。有这个系统时（用框架）：
你说：“我要一桌婚宴（需求）。”
系统自动：安排切菜工、炒菜师傅、摆盘员、传菜员协同工作，你只需要盯着出品质量。
结果：能稳定、高效地做出满汉全席（生产级复杂应用）。
LangChain：核⼼概念就是“链”（Chain），它将不同的模块（LLM, Prompts, Tools, Memory, Output Parsers）像乐⾼积⽊⼀样组合起来，构建复杂的 AI ⼯作流。“框架思维”驾驭“Vibe工具”，才是未来开发者最强的核心竞争力。
LangChain：LLM 应用开发的核心框架
LLM 驱动的应用程序的框架
目标是提供构建复杂 LLM 应用（如带有记忆的代理、复杂的RAG 系统、多步骤工作流）所需的全套工具。
维度 Python (LangChain) Java (LangChain4j) C++ (llama.cpp) 角色 AI应用框架核心 企业集成桥梁 推理加速引擎 功能完整度 ★★★★★ (100%) ★★★★☆ (~80%) ★★☆☆☆ (仅推理) 学习门槛 低（Python生态熟悉） 中（需理解AI概念） 高（C++ + 模型优化） 生产就绪度 需自建权限/队列等 开箱即用 推理稳定，但需包装 硬件依赖 建议GPU 可CPU/GPU CPU高效运行 C++ ⽣态在 LLM 应⽤开发的全栈框架领域相对缺失，但这背后有深刻的技术、⽣态和商业原因，主要 原因是因为 C++ 的定位不同
如何选择？
对于框架的选择，核⼼逻辑有以下⼏点：团队技术栈：
优先选择团队最熟悉的语⾔⽣态，以降低开发和学习成本。如果你所在的团队和技术栈是 Java，想快速为企业应⽤添加 AI 功能，LangChain4j 和 Spring AI 是绝佳的选择。如果你的需求是极致性能、离线运⾏或在 资源受限的环境（如⼿机、嵌⼊式设备）中部署模型，那么 C++ ⽣态的 llama.cpp 是你的不⼆之选。在⼤多数情况下，⼀个混合架构也很常⻅：例如，⽤ C++ 实现⾼性能推理引擎，然后⽤ Java 或Python 构建业务层和 API 接⼝。项⽬需求：
如果项⽬是研究性质或需要最⼤灵活性，Python（LangChain） 是不⼆之选。如果项⽬是以 RAG 为核⼼，LlamaIndex（Python/TS）提供了更专业的⼯具。如果项⽬需要深度集成到现有企业级后端（如 Spring 应⽤），则选择对应的 Spring AI。如果项⽬是⾯向 Web 的全栈应⽤或边缘函数，LangChain.js 是最佳选择。社区与⽀持：
Python 和 JS ⽣态的框架（LangChain）更新最快、社区最活跃，遇到问题更容易找到解决⽅案，是⼤多数⼈的选择。
LLM的问题
构建AI应用系统———>接入LLM————>LangChain
主流的六大使用原生LLM的场景问题（原生LLM在生产落地中面临的六大核心问题）
1. 幻觉（Hallucination）
问题本质：LLM本质是概率模型，它在"编造"看起来合理的内容，而非"回忆"事实。
具体表现：
捏造不存在的文献、法律条文、历史事件
错误引用公司内部文档（明明没写的内容，它说有）
数学计算错误、逻辑矛盾
解决思路：
RAG（检索增强生成）强制LLM基于外部知识回答
约束解码（如JSON模式、正则约束）
多模型交叉验证（让不同模型互查）
2. 指令遵循不稳定（Instruction Following Instability）
问题本质：同样的Prompt，同样的输入，LLM的输出在不同时间有细微差异。
具体表现：
要求输出JSON，有时输出纯文本
要求只输出"是/否"，有时输出解释
复杂指令（8条以上）必然遗漏某条
解决思路：
结构化Prompt（XML标签、Markdown格式）
温度参数降到0（降低随机性）
用验证层（Pydantic解析失败则重试）
引入少量示例（Few-shot）
3. 上下文记忆衰减（Lost in the Middle）
问题本质：LLM对输入内容的注意力分布不均，对开头和结尾的记忆好，中间部分容易"遗忘"。
具体表现：
给LLM 20条产品信息，让它检索某条，中间的第10条常被忽略
长对话中，第5轮提到的信息到第20轮就"消失"了
长文档QA，答案在文档中间时准确率急剧下降
解决思路：
关键信息放到Prompt开头或结尾
使用RAG而非把整个文档塞进上下文
定时总结对话历史（Context Compression）
4. 偏见与有害输出（Bias & Toxicity）
问题本质：训练数据中隐含的社会偏见、刻板印象被LLM学习并放大。
具体表现：
职业性别偏见（"护士"默认用"她"，"CEO"默认用"他"）
种族、地域刻板印象
可能输出危险内容（如制作危险品的步骤）
解决思路：
内容安全审核（接入内容审核API）
系统Prompt约束（强制要求"回答需要客观中立"）
模型微调（用去偏数据集）
5. 数值与逻辑推理弱（Weak Reasoning）
问题本质：LLM没有真正的"计算"和"逻辑推理"能力，它只是在模仿见过的推理模式。
具体表现：
简单算术错误（"1.1万 * 0.8 = 8800"这种都算错）
多步推理中某一步出错导致全盘错
无法真正理解"如果A则B，非B，所以非A"这样的逻辑
解决思路：
不依赖LLM做数值计算（让它写代码执行，或用计算器工具）
Chain-of-Thought（展示推理步骤）
外部验证（让LLM写出推理过程，再用代码验证）
6. 数据隐私与合规风险（Privacy & Compliance）
问题本质：调用API时，数据会传给第三方模型提供商，且可能被用于模型训练。
具体表现：
把客户PII（手机号、身份证）发给OpenAI → GDPR违规
把公司核心代码发给云模型 → 商业机密泄露
某些行业（金融、医疗）要求数据不能出境
解决思路：
本地部署开源模型（Llama 3、Qwen等）
数据脱敏后发送（替换敏感字段为占位符）
私有化部署（VPC内）
签署数据处理协议（限制模型提供商不能使用数据训练）
六大问题速览表
问题 核心表现 短期解决方案 根本解决需要 幻觉 捏造事实 RAG + 约束解码 更好的真实世界理解 指令不稳定 同Prompt不同输出 温度=0 + 验证层 确定性的推理框架 记忆衰减 忘记上下文中间内容 关键信息放开头/结尾 注意力机制改进 偏见与毒性 输出歪曲/有害内容 审核过滤 + 微调 训练数据清洗 逻辑推理弱 数学/逻辑错误 让LLM写代码而非计算 符号推理能力 隐私合规 数据泄露风险 本地部署 + 脱敏 全链路加密 7.供应商锁定 / 模型强耦合
问题本质：代码直接调用特定模型的API（如OpenAI的
chat.completions格式），换模型需要重写所有调用逻辑。解决方案：
LangChain抽象层：统一接口（
model.invoke()），底层可切换OpenAI兼容接口：很多开源模型（vLLM、LocalAI）提供OpenAI格式API
8.静态知识 / 无法自动更新
问题本质：模型训练完成后，知识就冻结在某个时间点，无法知道之后发生的新事件。
解决方案：
RAG：从实时数据库/搜索引擎检索最新信息
Fine-tuning + 定期更新：定时用新数据微调模型
联网搜索工具：让模型调用搜索API
LangChain
Releases · langchain-ai/langchain
定义
LangChain 是一个开源的编排框架，专门用来把大语言模型（如 GPT-4）与外部数据、工具、记忆、业务逻辑等组件“粘”在一起，从而构建出能理解上下文、自主推理和行动的智能应用。
通俗易懂举例
可以把大模型想象成一个智商超高、但只会张口就答的“大脑”，它没有记忆，也不会上网。
LangChain 就像给这个大脑装上了 记事本（记忆）、说明书和资料库（外部知识）、双手（调用搜索、计算器、API 等工具），让它能完成具体工作。
比如你让它：“总结昨天开会的 PDF，把结论发邮件给张三。” 用 LangChain 就能串联：读取 PDF → 提取文本 → 让模型总结 → 调用邮件工具发给指定联系人，全程自动完成。能干什么
带记忆的多轮对话：记住聊天历史，实现像真人客服一样的长对话。
检索增强生成（RAG）：先搜索本地文档、数据库或网页，再把搜到的知识喂给模型，让回答有据可查（如企业知识库问答）。
自主代理（Agent）：让模型自己决定用什么工具、按什么步骤解决问题，例如“帮我查明天北京天气，如果下雨就设置提醒带伞”。
链式调用（Chain）：把多个处理步骤串起来，比如“翻译成英文→提取关键词→生成标签”。
结构化数据提取：从一大段文字里精准抽取出姓名、金额、日期等字段。
地位
LangChain 是当前大模型应用开发领域 事实上的标准框架 之一。它率先统一了 LLM 应用开发的关键抽象，催生出庞大的生态（LangSmith 调试追踪、LangServe 部署、LangGraph 多步骤控制流）。无论是个人开发者快速做原型，还是企业搭建生产级 AI Agent 系统，都绕不开它。虽然也有 LlamaIndex（偏重索引检索）、Semantic Kernel 等同类竞品，但 LangChain 的社区活跃度、集成广度和思想普及度让它稳居“LLM 应用粘合剂”的头部位置。
LangChain 框架的设计精髓在于以链式（Chain）的⽅式整合多个组件，从⽽构建出功能丰富的⼤语言模型应用。链式表是表示LangChain 允许将多个步骤或多个组件串联起来，⽆需各个组件各⾃完成其能力，而是⼀次性执行这个"链"上的所有流程！举⼀个最简单的例⼦，若我们想借助提⽰词完成⼀次对于 LLM 的提问，在 LangChain 中⾄少需要定义两个组件：提⽰词模板组件⼤模型组件借助此图
这相当于，提⽰词模板组件执⾏了⼀次，⼤模型组件也执⾏了⼀次。⽽对于链式执⾏来说，只需执⾏⼀次链即可：
LangGraph：面向复杂工作流的图式架构
LangGraph 是 LangChain ⽣态系统中晚些出现的⼀个框架，其诞⽣背景与⼤型语⾔模型应⽤⽇益增长的复杂性密切相关。随着开发者尝试构建更⾼级的 AI 代理和多轮对话系统，传统链式结构的局限性逐渐显现：• 链式流程通常是线性的、预先定义好的步骤，难以处理需要循环、分⽀或⻓期状态维护的复杂场景。• 此外，在构建多智能体协作、需要人工介入（Human-in-the-loop）或⻓时间运⾏的任务时，需要更灵活的⼯作流管理和状态持久化支持。其实本质就是一句话LangChain无法满足复杂应用开发了 有局限性了 需要新的了就是LangGraph定义
LangGraph 是一个基于有状态图的编排框架，专为构建循环、多角色、可控流程的大模型应用而生。它把应用的每一步抽象成图中的“节点”，把决策和跳转逻辑抽象成“边”，并且天然管理每一步的历史状态，让模型能在需要时回溯、循环、等待人工介入。
举例
你让 LangChain 做“写报告 → 检查报告 → 不合格重写 → 再检查直到通过”，纯链式写法会很难受，很容易掉进无限递归或状态丢失。
LangGraph 则像在地图上画流程：
节点A：写报告；
节点B：评估报告；
边：如果评估“不合格”，自动走回路回到节点A重写；如果“合格”，走出口结束。
整个过程中，所有版本和历史都留在“状态”里，不会丢失上下文。
再复杂一点，还能让“研究员”节点和“编辑”节点互相传球，循环打磨，甚至随时暂停，等人类点“确认”再继续。能干什么（及 LangChain 的局限对照）
持久有状态的循环工作流：客服场景里连续多轮搜集信息、验证、纠错，失败后回退重新提问。纯 LangChain 的 Runnable 流很难优雅地表达“跳回前面某步”。
多智能体协作：多个 AI 角色（规划者、执行者、评审者）相互传递任务，形成内部循环，直到产出一致结果。
人机协同回路：流程中插入“等待人工审批”节点，批准后继续，拒绝则流转到修订节点。
分支与动态路由：根据用户意图实时选择不同处理路径，且后续能合并回主干。
可靠的长时运行任务：图执行可以被持久化、中断后恢复，适合要跑几分钟甚至几小时的复杂分析生成。
LangGraph 是如何用“图”让流程循环起来的
┌───────────┐
│ 开 始 │
└─────┬─────┘
│
▼
┌───────────────┐
│ 写笑话 │ ← 节点：writer
│ (拿状态里 │ 功能：生成/重写笑话
│ 的尝试次数 │ 更新：joke文本, attempts+1
│ 和反馈) │
└───────┬───────┘
│
▼
┌───────────────┐
│ 评估笑话 │ ← 节点：evaluator
│ (检查长度和 │ 功能：判断是否合格
│ 是否有问号) │ 输出：条件路由信号
└───────┬───────┘
│
┌───────▼───────┐
│ 条件判断 │ ← 条件边（conditional edge）
│ pass / fail ? │
└───┬───────┬───┘
│ │
pass │ │ fail (不合格)
▼ ▼
┌──────────┐ ┌──────────┐
│ 结 束 │ │ 回到写笑话│ ──┐
└──────────┘ └──────────┘ │
│
┌─────────────┘
│ （循环回路，携带完整历史状态）
▼
┌───────────┐
│ 写笑话 │ ← 再次执行，用上一次的 attempts 和上下文
└───────────┘
... 重复直到 pass用经典 LangChain 做同样的事，流程通常是一根筋的直线，很难表达循环
LangChain 团队于 2024 年推出了 LangGraph 框架，旨在提供⼀种图结构的、状态化的⽅式来构建复杂的 AI 代理应⽤。LangGraph 最初作为 LangChain 0.1.0 版本的⼀部分被引⼊，标志着 LangChain 从链式架构向图式架构的扩展。
LangChain上手
此模块需要科学上网哈！！！
调用LLM的流程
接入并定义大模型->定义消息->调用大模型->输出结果
远程调用
1.申请 API key 并配置环境变量
2.定义大模型
pip install -U langchain-openai# 定义⼤模型 from langchain_openai import ChatOpenAI model = ChatOpenAI(model="gpt-5-mini")# 定义消息列表 from langchain_core.messages import HumanMessage, SystemMessage messages = [ SystemMessage(content="Translate the following from English into Chinese"), HumanMessage(content="hi!"), ]result = model.invoke(messages) print(result)# 定义str字符串输出解析器 from langchain_core.output_parsers import StrOutputParser parser = StrOutputParser() print(parser.invoke(result))通过上述步骤，⽆论是调⽤⼤模型，还是输出解析，我们发现，每次都调⽤了⼀个 invoke() ⽅法，最终才会得到我们想要的结果。对于 LangChain，它给我们提供了链式执⾏的能⼒，即我们只需要定义各个“组件”，将它们“链起来”，⼀次性执⾏即可得到最终效果。from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage from langchain_core.output_parsers import StrOutputParser # 定义⼤模型 model = ChatOpenAI(model="gpt-5-mini") # 定义消息列表 messages = [ SystemMessage(content="Translate the following from English into Chinese"), HumanMessage(content="hi!"), ] # 定义输出解析器 parser = StrOutputParser() # 定义链 chain = model | parser # 执⾏链 result = chain.invoke(messages) print(result)
LangChain 相关概念
Runnable 接口
定义
Runnable接口是 LangChain 所有组件的统一通信协议。无论是大模型、提示模板、检索器还是工具，只要实现了Runnable，就能用一套标准方法调用、组合、并发、流式输出和批处理，并且能用|管道符像搭积木一样串联。通俗易懂举例
把 LangChain 的各个组件想象成不同品牌的智能家电，而Runnable就是统一的三孔插座。无论是烤面包机（提示模板）、微波炉（模型）、还是冰箱（检索器），只要它们都接上这个插座，你就能用同一套开关逻辑控制它们，甚至能把它们排成一排，让电流依次通过。
比如：提示模板 | 大模型 | 输出解析器，电器们就自动协作，不需要为每个电器配不同的转接头。能干什么
统一调用：所有组件都有
.invoke(input)、.stream()、.batch()、.ainvoke()等方法，学习成本极低。管道串联：用
a | b将一个Runnable的输出自动传入下一个Runnable的输入，形成“链”。自动并行/批处理：对列表输入自动分批、可配置并发，适合大批量数据处理。
流式透传：从第一个组件到最后一个组件都能流式输出，中间过程无阻塞。
中间件机制：可以通过
with_fallbacks()添加容错，bind()工具，config传参，非常灵活。怎么用
from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 这两个都是 Runnable prompt = ChatPromptTemplate.from_template("给我一个关于{topic}的冷笑话") model = ChatOpenAI(model="gpt-4o") # 用 | 管道符串成一条链，产生一个新的 Runnable chain = prompt | model # 统一使用 invoke result = chain.invoke({"topic": "程序员"}) print(result.content)地位
Runnable是 LangChain 表达“链式思维”的核心基石，它用极小的接口统一了所有模块，让整个生态能任意拼接。
但它本质上是无状态的、单向的流水线模型。当应用需要循环、回溯、人机协同、复杂状态管理时，Runnable的“链式”抽象就捉襟见肘了——比如很难在管道中优雅地实现“写→评→不合格重写”的回环。
这正是 LangGraph 诞生的技术动机：LangGraph 用“图”接管了Runnable管道的调度层，补充了回路、状态和分支等能力，但底层节点很多时候本身仍是Runnable组件，所以两者是共生关系。Runnable 定义了⼀个标准接⼝，允许 Runnable 组件：Invoked（调⽤）: 单个输⼊转换为输出Batched（批处理）: 多个输⼊被有效地转换为输出。Streamed（流式传输）: 输出在⽣成时进⾏流式传输。Inspected（检查）: 可以访问有关 Runnable 的输⼊、输出和配置的原理图信息。spected（检查）: 可以访问有关 uable 的输⼊、输出和配置的原理图信息。Composed（组合）: 可以组合多个 Runnable，以使⽤ LCEL 协同⼯作以创建复杂的管道。
LangChain Expression Language
定义
LangChain Expression Language（LCEL）是 LangChain 中用于声明式构建处理链的轻量级语言。它利用|（管道）操作符将多个Runnable组件串联成流水线，并自动处理并行、流式、异步、回退等高级行为。本质上，LCEL 就是LangChain 打造的一种“组件组合语法”。通俗易懂举例
想象 Unix 终端的管道命令：cat 食材 | 清洗 | 切配 | 下锅 | 装盘
每一步都接收上一步的输出，处理后又传给下一步。LCEL 正是把这个思想搬到了 AI 应用里prompt | model | parser
把提示模板、大模型、输出解析器像管道一样接起来，数据在中间自动流转，无需写一堆胶水代码。能干什么
极简链表达：一行代码描述复杂处理流，比如
(检索 + 提示) | 模型 | 解析器。自动并行：多个步骤如果没有依赖，LCEL 会自动并发执行，提高效率。
无缝流式传输：从链的起点到终点，所有组件都支持流式输出，延迟极低。
异步支持：用
ainvoke/astream就能一键切换异步，无需改动链结构。优雅容错：通过
.with_fallbacks()可添加备用模型，失败时自动切换。配置灵活：运行时可传入动态参数、工具绑定、自定义回调，而链定义不变。
怎么用
from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser # 定义组件（都是 Runnable） prompt = ChatPromptTemplate.from_template("把 {text} 翻译成英文") model = ChatOpenAI(model="gpt-4o") parser = StrOutputParser() # 用 LCEL 的 | 串起来 chain = prompt | model | parser # 普通调用 result = chain.invoke({"text": "你好，世界"}) print(result) # 输出: Hello, World # 流式输出 for chunk in chain.stream({"text": "今天天气真好"}): print(chunk, end="") # 添加回退 fallback_chain = chain.with_fallbacks([ChatAnthropic(model="claude-3-opus-20240229")])地位
LCEL 是 LangChain 整个生态的语法骨架，地位等同于 React 中的 JSX 或 SQL。它把组件组装从命令式编程提升为声明式配置，让代码极度简洁且易于维护。
它的底层完全依赖Runnable接口，所有能|连接的类都实现了Runnable。而Runnable又赋予了 LCEL 统一的行为（流式、异步、回退等），两者互为表里。
不过 LCEL 本身仍是无状态、无循环的管道模型，当应用需要记忆、分支回路、多代理协商时，就会暴露出抽象局限——而这恰好是 LangGraph 用“有状态图”补充的地方。但即便在 LangGraph 的节点内部，你仍然可以用 LCEL 构建执行子链，所以它不是被替代，而是向前兼容的基础层。
聊天模型
定义工具->绑定工具->调用工具
一般流程是
定义模型 定义工具 绑定工具 定义消息列表
结构化返回
在 LangChain 中，聊天模型提供了额外的功能：结构化输出。⼀种使聊天模型以结构化格式（例如 JSON）进⾏响应的技术。例如，可能希望将模型输出存储在数据库中，并确保输出符合数据库模式。这种需求激发了结构化输出的概念，其中可以指⽰模型使⽤特定的输出结构进⾏响应。with_structured_output()要想使⽤结构化输出能⼒，LangChain 提供了⼀种⽅法 .with_structured_output() ，该⽅法需要先定义输出结构，然后执⾏通过 .with_structured_output() 得到的 Runnable 实例。步骤如下（伪代码）：# 1. 定义输出结构 schema = {"foo": "bar"} # 2. 绑定schema，其实是⽣成⽀持结构化返回的 Runnable 实例 model_with_structure = model.with_structured_output(schema) # 3. 执⾏ structured_output = model_with_structure.invoke(user_input)这是获得结构化输出的最简单、最可靠的⽅法。此⽅法将 输出结构 作为参数输⼊，返回⼀个类似model 的 Runnable。不同之处在于执⾏ Runnable 后的输出结果，输出的不是字符串或消息 ，⽽是输出与给定输出结构相对应的对象。该输出结构 可以指定为 TypedDict 类、JSON Schema 或 Pydantic 类。如果使⽤ TypedDict 或JSON Schema，则 Runnable 将返回⼀个字典，如果使⽤ Pydantic 类，则将返回⼀个 Pydantic 对象。返回 Pydantic 对象
我们可以设置执⾏ Runnable 后的输出结果指定为 Pydantic 类，这将返回⼀个 Pydantic 对象当收到模型的响应后，LangChain 会提取出代表 Pydantic 参数的 JSON 对象，并⽤ Pydantic 模型对其进⾏解析和验证，将这个验证后的 JSON 转换为⼀个可⽤的 Pydantic 对象实例返回。如下所⽰： from langchain_openai import ChatOpenAI from typing import Optional from pydantic import BaseModel, Field # 定义⼤模型 model = ChatOpenAI(model="gpt-4o-mini") # 定义输出结构：Pydantic 类 class Joke(BaseModel): """给⽤⼾讲⼀个笑话。""" setup: str = Field(description="这个笑话的开头") punchline: str = Field(description="这个笑话的妙语") rating: Optional[int] = Field( default=None, description="从1到10分，给这个笑话评分" ) structured_model = model.with_structured_output(Joke) result = structured_model.invoke("给我讲⼀个关于唱歌的笑话") print(result) 返回 TypedDict
先了解⼀下 TypedDict ，它⽤于为字典对象提供精确的、结构化的类型提⽰。它允许我们指定⼀个字典中应该有哪些键，以及每个键对应的值的类型。最清晰、最常⽤的定义⽅式，就是类似于定义⼀个类（Python 3.8+），如下所⽰：from typing import TypedDict class User(TypedDict): name: str age: int email: str is_active: bool = True # 默认值因此，我们也可以设置执⾏ Runnable 后的输出结果指定为 TypedDict 类，这将返回⼀个字典，且输出后，会根据设定进⾏验证。以下是该⽅法的使⽤from langchain_openai import ChatOpenAI from typing import Optional from typing_extensions import Annotated, TypedDict # 定义⼤模型 model = ChatOpenAI(model="gpt-4o-mini") # 定义输出结构： TypedDict class Joke(TypedDict): """给⽤⼾讲⼀个笑话。""" setup: Annotated[str, ..., "这个笑话的开头"] punchline: Annotated[str, ..., "这个笑话的妙语"] rating: Annotated[Optional[int], None, "从1到10分，给这个笑话评分"] structured_model = model.with_structured_output(Joke) result = structured_model.invoke("给我讲⼀个关于唱歌的笑话") print(result)返回 JSON
还可以让聊天模型直接返回 JSON，只不过为了声明 JSON，我们需要定义 JSON Schema，如下from langchain_openai import ChatOpenAI # 定义⼤模型 model = ChatOpenAI(model="gpt-4o-mini") json_schema = { "title": "joke", "description": "给⽤⼾讲⼀个笑话。", "type": "object", "properties": { "setup": { "type": "string", "description": "这个笑话的开头", }, "punchline": { "type": "string", "description": "这个笑话的妙语", }, "rating": { "type": "integer", "description": "从1到10分，给这个笑话评分", "default": None, }, }, "required": ["setup", "punchline"], } structured_model = model.with_structured_output(json_schema) result = structured_model.invoke("给我讲⼀个关于唱歌的笑话") print(result)打印结果
{'setup': '为什么唱歌的⼈总是很开⼼？', 'punchline': '因为他们总是有很多⾳符可供选 择！', 'rating': 7}声明格式为JSON Schema 但是实际效果是字典
实用场景
作为信息提取器
from langchain_openai import ChatOpenAI from typing import Optional from pydantic import BaseModel, Field from langchain_core.messages import HumanMessage, SystemMessage # 定义⼤模型 model = ChatOpenAI(model="gpt-4o-mini") class Person(BaseModel): """⼀个⼈的信息。""" # 注意: # 1. 每个字段都是 Optional “可选的” —— 允许 LLM 在不知道答案时输出 None。 # 2. 每个字段都有⼀个 description “描述” —— LLM使⽤这个描述。 # 有⼀个好的描述可以帮助提⾼提取结果。 name: Optional[str] = Field(default=None, description="这个⼈的名字") hair_color: Optional[str] = Field(default=None, description="如果知道这个⼈头 发的颜⾊") skin_color: Optional[str] = Field(default=None, description="如果知道这个⼈的 肤⾊") height_in_meters: Optional[str] = Field(default=None, description="以⽶为单 位的⾼度") structured_model = model.with_structured_output(schema=Person) messages = [ SystemMessage(content="你是⼀个提取信息的专家，只从⽂本中提取相关信息。如果您不知道 要提取的属性的值，属性值返回null"), HumanMessage(content="史密斯⾝⾼6英尺，⾦发。") ] result = structured_model.invoke(messages) print(result)打印结果
name='史密斯' hair_color='⾦发' skin_color=None height_in_meters='1.83'使用"少样本提示"来增强信息提取能力
与工具结合
流式传输
定义
流式传输（Streaming）是一种边生成边返回的数据传输方式。模型不再等整个答案全部写好再一次性丢给你，而是每算出一个字、一个词，就立刻通过网络推送过来，像打印机一样逐行吐出内容。
通俗易懂举例
非流式（invoke）：像写信，对方写完装进信封再寄给你，你等好几天才看到全文。
流式（stream）：像微信聊天时对方“正在输入…”状态，你看着他的消息一个字一个字冒出来，不用等他全部打完。
打开 ChatGPT 网页版时，答案是一行一行动态出现的，那就是流式传输。能干什么
大幅降低用户感知延迟：用户看见第一个字出现的时间可以从几十秒缩短到毫秒级，体验极其流畅。
支持中途打断：如果生成方向跑偏，用户可以随时停止，不用浪费时间和算力等它生成完。
适应长文本生成：写文章、代码、翻译长篇内容时，流式输出让阅读体验像看直播，不会死等。
便于实时处理：前端可以边接收边逐句朗读（TTS）、逐行显示或逐块解析。
怎么用（LangChain 示例）
from langchain_openai import ChatOpenAI model = ChatOpenAI(model="gpt-4o") # 流式调用，返回一个生成器 for chunk in model.stream("给我讲个关于程序员的冷笑话"): print(chunk.content, end="", flush=True) # 逐字输出
stream() 同步传输
astream() 异步传输
定义
stream()和astream()都是 LangChain 中Runnable接口提供的流式调用方法，区别在于执行模型不同：
stream()：同步流式。代码会卡在当前线程，逐个等待模型生成下一个片段，直到全部生成完才继续往下走。
astream()：异步流式。使用async/await，在等待模型生成的过程中，程序可以去干别的事（如同时处理其他请求），不阻塞线程。通俗易懂举例
同步 stream()：像你去柜台点餐，店员给你做汉堡，你站在窗口一直等，每做好一层（面包、肉饼、生菜）就递给你一层，但你啥也干不了，只能全程干等。
异步 astream()：像你用餐厅小程序下单，下单后就坐回去玩手机、聊天，后厨做好一层就推送一条通知到你手机，你可以边接收边做别的事。
核心区别
对比维度 stream()astream()执行方式 同步阻塞 异步非阻塞 语法 for chunk in model.stream(...)async for chunk in model.astream(...)适用场景 单个对话、脚本、Jupyter Web服务器、高并发 API 服务 能否并发处理多请求 ❌ 同一时间只能等一个 ✅ 可同时处理多个请求 怎么用
同步 stream()
from langchain_openai import ChatOpenAI model = ChatOpenAI(model="gpt-4o") # 程序会卡在这里，直到模型全部生成完才退出循环 for chunk in model.stream("推荐三本书"): print(chunk.content, end="", flush=True) # 这段代码后面的逻辑要等循环结束才能执行异步 astream()
import asyncio from langchain_openai import ChatOpenAI model = ChatOpenAI(model="gpt-4o") async def main(): # 异步循环：在等待 chunk 的间隙，可以切换去执行其他任务 async for chunk in model.astream("推荐三本书"): print(chunk.content, end="", flush=True) # 如果同时有多个 main() 运行，它们不会互相阻塞 asyncio.run(main())为什么要区分？
如果只做本地测试、写脚本，stream()完全够用。
但当你用 FastAPI/Flask 搭建 AI 服务，同时有 100 个用户在线时，用astream()可以让服务器在等待模型生成时去处理其他人的请求，避免整个服务被卡死。这是生产环境必备的能力。补充小技巧
LangChain 还提供了ainvoke()（异步非流式调用）和abatch()（异步批处理），命名规则统一：
前面加
a→ 异步版本带
stream→ 流式版本组合：
astream= 异步 + 流式所以不管你搭建什么形态的应用，
Runnable接口都能提供对应的调用方式。总结⼀下：• 协程是 asyncio 的核⼼概念之⼀。它是⼀个特殊的函数，可以在执⾏过程中暂停，并在稍后恢复执⾏。协程通过 async def 关键字定义，并通过 await 关键字暂停执⾏，等待异步操作完成。• 要运⾏⼀个协程，可以使⽤ asyncio.run() 函数。它会创建⼀个事件循环，并运⾏指定的协程。事件循环是 asyncio 的核⼼组件，负责调度和执⾏协程。它不断地检查是否有任务需要执行，并在任务完成后调⽤相应的回调函数。
自定义流式输出
定义
自定义流式输出 是指在 LangChain 的流式传输过程中，插入你自己写的处理逻辑，让每一个新生成的文字片段（token）在输出的同时，再额外做你想做的事情——比如实时统计字数、敏感词过滤、动态高亮、同步朗读（TTS）、存入日志等。
通俗易懂举例
正常的stream()像水龙头直接流水，你只能看着水出来。自定义流式输出就是在水龙头上接一个透明过滤器：水流过的时候，你可以在过滤器里做染色、测水质、记录水流速，但水依然能顺畅流出，不影响原来接水的人。能干什么
实时防敏感：边生成边检查，一旦碰到违规词马上截断或替换。
同步朗读 (TTS)：模型吐出一个短句，你立刻转成语音播放。
打字机效果增强：前端逐字显示的同时，你可以统计字数、动态调整显示速度。
埋点与日志：记录每段 token 的生成时间，分析响应延迟。
流式后处理：比如把 Markdown 逐段转换成 HTML 再推送给前端。
怎么用
LangChain 提供了两种主流方式来实现自定义流式输出。方式1：通过 Callback 拦截
on_llm_new_token这是最常用也最简单的办法。你写一个回调类，继承
BaseCallbackHandler，重写on_llm_new_token方法：from langchain.callbacks.base import BaseCallbackHandler from langchain_openai import ChatOpenAI class MyStreamHandler(BaseCallbackHandler): def on_llm_new_token(self, token: str, **kwargs) -> None: # token 就是刚生成的最小片段（通常是一个词或一个字） print(f"[实时token: {token}]", end="", flush=True) # 在这里可以：敏感词检测、字数累加、TTS 等 if "敏感词" in token: print("<拦截>", end="") # 把自定义 handler 传给模型 model = ChatOpenAI( model="gpt-4o", streaming=True, # 必须开启流式 callbacks=[MyStreamHandler()] ) # 现在调用 invoke 也会自动触发回调 model.invoke("讲个笑话")注意：即使你用
invoke（非流式调用），只要模型构造时设了streaming=True且挂载了回调，on_llm_new_token也会被触发，但你无法获取最终汇总结果（那要用链式调用配合）。更典型的用法是直接在.stream()基础上包装生成器：方式2：包装生成器（更灵活）
直接对
stream()返回的生成器做二次包装，这样你可以拿到每个 chunk 并处理，同时保留继续传递给下游的能力。def custom_stream(prompt: str): model = ChatOpenAI(model="gpt-4o") total_chars = 0 for chunk in model.stream(prompt): text = chunk.content total_chars += len(text) # 自定义处理：比如过滤脏话 clean_text = text.replace("糟糕词", "***") # 用 yield 把处理后的内容传出去，保持生成器形式 yield clean_text print(f"\n总字数：{total_chars}") # 使用 for processed_chunk in custom_stream("讲个故事"): print(processed_chunk, end="")
深度探索流式传输
SSE 协议介绍
HTTP 协议本⾝设计为⽆状态的请求-响应模式，严格来说，是⽆法做到服务器主动推送消息到客⼾端，但通过 Server-Sent Events （服务器发送事件，简称 SSE）技术可实现流式传输，允许服务器主动向浏览器推送数据流。也就是说，服务器向客⼾端声明，接下来要发送的是流消息(streaming)，这时客⼾端不会关闭连接，会⼀直等待服务器发送过来新的数据流。SSE（Server-Sent Events）是⼀种基于 HTTP 的轻量级实时通信协议，浏览器可以通过内置的EventSource API 接收并处理这些实时事件核心特点•基于 HTTP 协议复⽤标准 HTTP/HTTPS 协议，⽆需额外端⼝或协议，兼容性好且易于部署。•单向通信机制SSE 仅⽀持服务器向客⼾端的单向数据推送，客⼾端通过普通 HTTP 请求建⽴连接后，服务器可持续发送数据流，但客⼾端⽆法通过同⼀连接向服务器发送数据。•⾃动重连机制⽀持断线重连，连接中断时，浏览器会⾃动尝试重新连接（⽀持 retry 字段指定重连间隔）。•⾃定义消息类型客⼾端发起请求后，服务器保持连接开放，响应头设置 Content-Type: text/event/stream ，标识为事件流格式，持续推送事件流。
为什么大模型场景选 SSE 而不是 WebSocket？
对比维度 SSE WebSocket 通信方向 单向（服务器→客户端） 双向 协议复杂度 极简，纯文本格式 需要握手和帧协议 防火墙兼容性 标准 HTTP，无需额外端口 可能被企业防火墙拦截 断线重连 内置 id+retry机制需手动实现 大模型场景是典型的单向数据流：客户端发送提示词后，只需要接收数据。SSE 用标准 HTTP 端口，配置工作几乎没有，而且内置了断线重连机制（通过
id字段和retry字段）。一句话总结底层原理：LangChain 的流式传输，本质是对 LLM 服务商 SSE 分块响应的封装——你看到的是一个 Python 迭代器，底层跑的是 HTTP 分块传输，来来去去的是一行行
data。
说白了 sse单项通信机制比较合适这种情况
LangChain 如何支持流式传输？
开始我们就说了，LangChain 本⾝并不“创造”或“规定”⼀个底层的⽹络传输协议，⽽是依赖于其底层的⼤模型供应商（如 OpenAI）的协议。因此当我们发起请求时，会在请求中设置 stream=True （ _stream() 源码中的第⼀步），表示OpenAI 服务器将在⽣成 Response 时向客⼾端发出数据（server-sent events，SSE）。此时 API会保持 HTTP 连接打开，并以特定格式发送数据。
总结⼀下：1. langchain-openai 包通过集成 OpenAI Python SDK，提供了⼀个 HTTP 客户端。2. 因此，⽀持 LangChain 向 OpenAI 的 API 发起调用请求。3. 若希望发起流式传输请求，则需在请求中加入 stream=True ，向 OpenAI 说明以 SSE 协议进行流式返回。4. LangChain 接收 OpenAI 的 SSE 格式的响应，并将其转换为 LangChain ⾃封装的消息格式，如AIMessageChunk 消息。这样就可以以统⼀的⽅式处理来⾃不同模型提供商（OpenAI, Anthropic等）的流式响应。
核心组件
消息（Messages）
消息是聊天模型中的通信单位，用于表示聊天模型的 输入 和 输出 ，以及可能与对话关联的任何其他上下文或元数据 。
LLM 消息结构
大语言模型（LLM）的 API 通常接受一个由多条消息组成的数组作为输入，每条消息包含两个基本属性：
role（角色）：标识消息的来源，常见值有
system（系统提示）、user（用户）、assistant（AI 回复）。content（内容）：消息的实际文本。
例如 OpenAI 的消息结构：
json [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "What is LangChain?"}, {"role": "assistant", "content": "LangChain is a framework..."} ]这种结构天然支持多轮对话：将历史消息拼接后发给 LLM，模型根据上下文生成下一轮回复。
LangChain 消息
LangChain 对 LLM 原生消息结构进行了抽象与封装，提供统一的接口，以便兼容不同模型提供商（OpenAI、Anthropic、Cohere 等）。
BaseMessage 抽象消息类
BaseMessage是所有消息类型的抽象基类，定义了通用属性和方法：
属性：
content：消息文本内容。
type：消息类型标识（如"human"、"ai"、"system"、"function"等）。
additional_kwargs：用于存储模型特定的附加信息（如 OpenAI 的function_call）。常用子类：
HumanMessage：对应user角色。
AIMessage：对应assistant角色。
SystemMessage：对应system角色。
FunctionMessage/ToolMessage：用于函数调用场景。使用示例：
from langchain_core.messages import HumanMessage, AIMessage messages = [ SystemMessage(content="You are a helpful assistant."), HumanMessage(content="什么是LangChain？"), AIMessage(content="LangChain是一个用于开发LLM应用的框架。") ]通过统一的
BaseMessage接口，LangChain 的链（Chain）和代理（Agent）可以无差别地处理不同来源的消息。缓存历史消息
多轮对话
只要将历史消息，重新发送给聊天模型，那么就可以实现多轮对话的功能内存缓存
那么对于历史消息的管理就显得尤为重要。在 LangChain ⽼版本中，可以RunnableWithMessageHistory 消息历史类来包装另⼀个 Runnable 并为其管理聊天消息历史记录。它将跟踪模型的输⼊和输出，并将其存储在某个数据存储中。未来的交互将加载这些消息，并将其作为输⼊的⼀部分传递给链。from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage from langchain_core.chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory from langchain_core.runnables.history import RunnableWithMessageHistory # 定义⼤模型 model = ChatOpenAI(model="gpt-4o-mini") store = {} # 接受⼀个 session_id 并返回⼀个消息历史对象。 # 这个 session_id ⽤于区分不同的对话，并应作为配置的⼀部分在调⽤新链时传⼊ def get_session_history(session_id: str) -> BaseChatMessageHistory: if session_id not in store: # InMemoryChatMessageHistory() 将消息存储在内存列表中。 store[session_id] = InMemoryChatMessageHistory() return store[session_id] # 包装model，管理聊天消息历史记录 with_message_history = RunnableWithMessageHistory(model, get_session_history) config = {"configurable": {"session_id": "1"}} with_message_history.invoke( [HumanMessage(content="Hi! I'm Bob")], config=config, ).pretty_print() with_message_history.invoke( [HumanMessage(content="What's my name?")], config=config, ).pretty_print()从 LangChain 的 v0.3 版本开始，官⽅建议 LangChain ⽤⼾不要使⽤RunnableWithMessageHistory ，⽽是利⽤ LangGraph 持久性 来完成（见 LangGraph 章节）。原因是它们的功能有限，不太适合现实世界的对话式 AI 应⽤程序。这些内存抽象缺乏对多用户、多对 话场景的内置⽀持，⽽这对于实际的对话式⼈⼯智能系统⾄关重要。这些实现中的⼤多数已在LangChain 0.3.x 中被正式弃⽤，取⽽代之的是 LangGraph 持久性 。 LangGraph 持久性 ⾮常灵活，可以⽀持⽐ RunnableWithMessageHistory 接⼝更⼴泛的⽤例。我们会在 LangGraph 篇章中学习它！因此， RunnableWithMessageHistory 这部分我们讲解的并不深⼊。在之前，对于⽣产环境我们还需要使⽤聊天消息历史记录的持久化实现，例如 RedisChatMessageHistory （） ，⽽不是 InMemoryChatMessageHistory() ，但现在也已不推荐新应⽤使⽤它们了。管理历史消息
随着对话轮次增多，消息列表会变得很长，导致 token 超限、成本上升、响应延迟。因此需要主动管理历史消息，主要包括：裁剪、过滤、合并。
上下文窗口
每个 LLM 模型能同时处理的最大 token 数（输入 + 输出）称为上下文窗口。例如：
GPT-3.5-turbo 窗口为 16K token
GPT-4-32K 为 32K token
Claude 3 某些版本可达 200K token
一旦消息总 token 数超过窗口，模型会截断最早的 token，导致丢失重要信息。因此必须确保发送给模型的消息总长度不超过窗口限制。
Token
Token 是⽂本的基本单位。它不是完全等同于⼀个单词或⼀个汉字，是⼀个更细粒度的划分。Token 是 LLM 处理文本的最小单位。一个 token 大约对应 0.75 个英文单词，或一个汉字（不同分词器规则不同）。计算 token 数可使用模型对应的分词器（如 OpenAI 的
tiktoken）。LangChain 也提供了get_num_tokens()等辅助方法。消息裁剪 (Truncation)
裁剪是指从消息列表中移除部分消息，以降低总 token 数。LangChain 提供了
trim_messages工具（位于langchain_core.messages）来灵活裁剪输⼊ = 系统消息 + 对话历史 + 最新⽤户问题对于模型来说，并不真正“记忆”，⽽是每次都将完整的上下⽂重新输⼊基于输入 Token 数的修剪
指定一个最大 token 数（例如 4000），
trim_messages会从消息列表的最早位置开始丢弃消息，直到剩余消息的 token 总数小于等于该限制。同时可以保留一定数量的最新消息（例如始终保留最后 2 条用户-助手轮次），避免完全丢失近期上下文。示例：
from langchain_core.messages import trim_messages trimmed = trim_messages( messages, max_tokens=4000, strategy="last", # 保留最后的消息 token_counter=my_token_counter, # 自定义或使用模型默认的计数器 include_system=True, # 始终保留系统消息 allow_partial=False, # 若一条消息超出限制则直接丢弃（一般设为False） )基于消息数的修剪
更简单的策略：直接限制消息条数。例如只保留最近的 10 条消息。LangChain 的
trim_messages也支持max_messages参数：trimmed = trim_messages( messages, max_messages=10, # 最多保留10条消息 strategy="last" )注意：基于条数的裁剪无法精确控制 token 数，因为每条消息的长度差异很大。推荐优先使用基于 token 的修剪。
消息过滤
过滤是指根据某些条件选择性移除消息，而不是单纯按位置或长度裁剪。常见场景：
移除包含特定关键词的消息（如敏感信息）。
只保留
HumanMessage和AIMessage，丢弃SystemMessage（某些后续处理不需要系统提示）。过滤掉工具调用产生的中间消息。
实现过滤通常使用列表推导式或自定义函数：
filtered = [m for m in messages if not isinstance(m, ToolMessage)]LangChain 没有提供专门的
filter_messages工具，但可以结合trim_messages的自定义filter_func参数（部分版本支持）或直接手动过滤。】在更复杂的场景下，我们可能会使⽤消息列表来跟踪状态，例如我们可能只想将这个完整消息列表的⼦集传递模型调⽤，⽽不是所有的历史记录。filter_messages ⽅法则可以轻松地按类型、ID 或名称过滤 message。准备消息列表from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, filter_messages # 历史消息记录 messages = [ SystemMessage("你是⼀个聊天助⼿", id="1"), HumanMessage("⽰例输⼊", id="2"), AIMessage("⽰例输出", id="3"), HumanMessage("真实输⼊", id="4"), AIMessage("真实输出", id="5"), ]筛选
print(filter_messages(messages, include_types=[HumanMessage, AIMessage], exclude_ids=["3"])) # 结果如下： # [ # HumanMessage(content='⽰例输⼊', additional_kwargs={}, response_metadata= {}, name='example_user', id='2'), # HumanMessage(content='真实输⼊', additional_kwargs={}, response_metadata= {}, name='bob', id='4'), # AIMessage(content='真实输出', additional_kwargs={}, response_metadata={}, name='alice', id='5') # ]消息合并
合并是指将多条消息的内容拼接成一条，以减少消息数量。例如将连续的多条用户消息合并为一条，或者将连续多轮对话压缩成摘要。常见的实现方式：
简单拼接：将多个
HumanMessage的 content 用换行符连接，生成一条新的HumanMessage。摘要合并：使用 LLM 将一段历史对话生成一个简短的摘要，替代原始的多条消息。
LangChain 的
ConversationSummaryMemory就是合并的典型应用：它会定期调用 LLM 生成历史摘要，并将摘要作为系统消息保留，而原始消息被丢弃，从而大幅压缩 token 使用。示例（手动合并）：
def merge_consecutive_user_messages(messages): new_messages = [] for msg in messages: if isinstance(msg, HumanMessage) and new_messages and isinstance(new_messages[-1], HumanMessage): new_messages[-1].content += "\n" + msg.content else: new_messages.append(msg) return new_messages若我们的消息列表存在连续某种类型相同的消息，但实际上某些模型不⽀持传递相同类型的连续消息。因此对于这种情况，我们可以使⽤ merge_message_runs ⽅法轻松合并相同类型的连续消息。例如下：from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, merge_message_runs # 定义⼤模型 model = ChatOpenAI(model="gpt-4o-mini") # 历史消息记录 messages = [ SystemMessage("你是⼀个聊天助⼿。"), SystemMessage("你总是以笑话回应。"), HumanMessage("为什么要使⽤ LangChain?"), HumanMessage("为什么要使⽤ LangGraph?"), AIMessage("因为当你试图让你的代码更有条理时，LangGraph 会让你感到“节点”是个好主 意！"), AIMessage("不过别担⼼，它不会“分散”你的注意⼒！"), HumanMessage("选择LangChain还是LangGraph?"), ] merged = merge_message_runs(messages) # 打印合并后的每个消息 print("\n".join([repr(x) for x in merged]))合并结果
SystemMessage(content='你是⼀个聊天助⼿。\n你总是以笑话回应。', additional_kwargs= {}, response_metadata={}) HumanMessage(content='为什么要使⽤ LangChain?\n为什么要使⽤ LangGraph?', additional_kwargs={}, response_metadata={}) AIMessage(content='因为当你试图让你的代码更有条理时，LangGraph 会让你感到“节点”是个好 主意！\n不过别担⼼，它不会“分散”你的注意⼒！', additional_kwargs={}, response_metadata={}) HumanMessage(content='选择LangChain还是LangGraph?', additional_kwargs={}, response_metadata={})调用大模型
## ⽅式—： merged = merge_message_runs(messages) model.invoke(messages).pretty_print() ## ⽅式⼆： merger = merge_message_runs() chain = merger | model chain.invoke(messages).pretty_print()打印结果
================================== Ai Message ================================== 这就像选择汉堡还是热狗！如果你想要⼀个多层次的体验，选择LangChain；如果你想要⼀个清晰的链 接，LangGraph就像是你的“⽀架”！让你的代码加个“料”！
提示词模板
提⽰词模板（Prompt Template）是 LangChain 的核⼼抽象之⼀，它被⼴泛应⽤于构建⼤语⾔模型（LLM）应⽤的各个环节。简单来说，只要是需要动态、批量、或有结构地向⼤语⾔模型【发送请求】的地⽅，⼏乎都会⽤到提⽰词模板字符串模板
聊天消息模板
消息占位符
使⽤ LangChain Hub 的提⽰词模板
LangChain Hub 是⼀个⽤于上传、浏览、拉取和管理提⽰词(prompts)的地⽅。随着 LLM 的发展，提⽰变得越来越重要。LangChain 正在打造⼀个与像 GitHub 这样的传统平台，GitHub⻓期以来⼀直是共享和协作代码的⾸选平台。于是推出了 LangChain Hub 平台。LangChain Hub 创建⼀个分享和发现 Prompt 的平台，使得开发者可以更容易地发现新⽤例和精炼提⽰。 这⼀举措使提⽰⼯程师更容易合作，重复使⽤现有的提⽰，并对其进⾏微调以实现特定的结果，从⽽加速对话代理和其他基于语⾔的应⽤程序的开发和部署。早期的时候 LangChain Hub 有Prompt、Chain、Agent，现在只有Prompt。
少样本提示
少样本提⽰是⼀种通过向 LLM 提供少量具体⽰例或样本，来教会它如何执⾏某项特定任务的技术。提⾼模型性能的最有效⽅法之⼀是给出⼀个【模型⽰例】指导⼤模型你想做什么、怎么做。下⾯⽤例⼦解释少样本提⽰的作⽤。少样本提⽰则是在给出考题前，先给它看⼏道类似的、附有正确答案的例题。添加⽰例输⼊和预期输出的技术给到模型提⽰，让模型通过例题来理解任务应该怎么做。这能解决什么问题？LLM 虽然知识渊博，但有时我们需要它以⾮常特定的格式、⻛格或逻辑来回答问题。提供正确的⽰例可以减少模型“胡说⼋道”或犯低级错误的概率，将其输出约束在你提供的范例范围内。举个例⼦更能理解：1. 强制要求模型以特定的格式（如JSON、XML、特定的列表样式）输出结果。样例可以当作格式样板。2. 有些任务很难⽤⽂字指令清晰描述（例如：“请⽤莎⼠⽐亚的⻛格写作”）。提供⼏个例⼦⽐写⻓篇⼤论的指令更有效。3. 对于需要多步推理的复杂任务，⽰例可以展⽰出思考链，引导模型遵循类似的推理路径。实现少样本提⽰
实现少样本提⽰的第⼀步也是最重要的⼀步是提出⼀个好的⽰例数据集。好的⽰例应该在运⾏时相关、清晰、信息丰富，并提供模型尚不知道的信息。如我们给出的例⼦，就能很好的提⽰⼤模型 🦜 与 ➕ 含义相似examples = [ {"input": "2 🦜 2", "output": "4"}, {"input": "2 🦜 3", "output": "5"}, ]如何让⼤模型看懂这份⽰例呢？之前我们说过聊天模型读的是聊天消息。因此，接下来我们需要将⽰例集实例化成聊天模型可以读懂的聊天消息。对于 LangChain 就需要创建⼀个FewShotChatMessagePromptTemplate 对象来实例化⽰例集。FewShotChatMessagePromptTemplate 是⼀个提⽰词模板，专⻔⽤来将⽰例集实例化为聊天消息，⽤法如下所⽰：from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate example_prompt = ChatPromptTemplate( [ ("human", "{input}"), ("ai", "{output}"), ] ) few_shot_prompt = FewShotChatMessagePromptTemplate( example_prompt=example_prompt, # ChatPromptTemplate，⽤于格式化单个⽰例 examples=examples, # 样本⽰例 ) print(few_shot_prompt.invoke({}).to_messages())[ HumanMessage(content='2 🦜 2', additional_kwargs={}, response_metadata= {}), AIMessage(content='4', additional_kwargs={}, response_metadata={}), HumanMessage(content='2 🦜 3', additional_kwargs={}, response_metadata= {}), AIMessage(content='5', additional_kwargs={}, response_metadata={}) ]最后，得到⽰例集消息列表后，就可以带上⼀起发起请求：final_prompt = ChatPromptTemplate( [ ("system", "你是⼀个神奇的数学奇才。"), few_shot_prompt, ("human", "{input}"), ] ) from langchain_openai import ChatOpenAI model = ChatOpenAI(model="gpt-4o-mini") chain = final_prompt | model chain.invoke({"input": "What is 2 🦜 9?"}).pretty_print()结果如下================================== Ai Message ================================== 11推理引导
使⽤⽰例数据增强 LangChain 信息提取能⼒
⽰例选择器
⼀旦我们有了⽰例数据集，就需要考虑提⽰中应该有多少个⽰例。关键的权衡是，更多的⽰例通常会提⾼性能，但更⼤的提⽰会增加成本和延迟。超过某个阈值，太多⽰例可能会开始混淆模型。找到正确数量的⽰例在很⼤程度上取决于模型、任务、⽰例的质量以及成本和延迟限制。有趣的是，模型越好，它需要精准的⽰例就越少。但其实，最佳的⽅法是使⽤不同数量的⽰例进⾏⼀些实验。若此时我们有【⼤量】的⽰例数据集。对于⼤模型来说，就没必要全部使⽤与参考。我们需要有⼀种⽅法可以根据给定的输⼊，从数据集中选择⽰例。在 LangChain 中，⽰例选择器就可以帮我们从⼀组【⽰例的集合】中根据具体策略选择正确的【⽰例⼦集】构建少样本提⽰。
按⻓度选择⽰例
当我们担⼼构造提⽰时，将超过上下⽂窗⼝⻓度，根据特定⻓度内可以容纳的数量选择⽰例。对于较⻓的输⼊，它将选择更少的⽰例来包含；⽽对于较短的输⼊，它将选择更多⽰例。按语义相似性选择⽰例
什么是语义相似？它是衡量⽂本在【含义上】的接近程度。例如下述两段⽂本：text1 = "我喜欢猫"text2 = "我讨厌狗"这两段⽂本表⾯相似度低，但语义上都是表达对动物的态度。再例如：text1 = "苹果很甜"text2 = "苹果市值创新⾼""苹果"可以指⽔果或公司，语义相似可以解决⼀词多义问题，因此这两段⽂本语义上不相似。LangChain 能根据输⼊和⽰例之间的语义相似性来决定选择哪些⽰例，它通过查找与输⼊具有最⼤余弦相似性的嵌⼊⽰例来实现这⼀点。按最⼤边际相关性选择⽰例（MMR）
什么是最⼤边际相关性？它是⼀种重新排序算法，它使⽤语义相似性作为基础⼯具，从⼀个候选集中挑选出⼀组既能代表查询主题⼜彼此多样化的结果。听起来好像和语义相似性类似，⽤⼀个例⼦看下两者的区别：• 【语义相似性】就像⾯试官衡量每个应聘者与职位要求的匹配度。他会给每个应聘者打⼀个分数。• 【最⼤边际相关性】就像团队经理（MMR算法）要组建⼀个团队。⽬标是选出⼀组“精华”结果⽽不是⼀个单⼀结果：每个成员都要满⾜基本职位要求（满⾜相关性）。但经理不希望团队⾥全是只会⼀种技能的程序员。他需要前端、后端、算法、测试等不同专⻓的⼈，以确保团队能⼒全⾯、减少冗余（新颖性/多样性）。经理的策略是：先招⼀个最匹配的技术⼤⽜（第⼀步），然后接下来招的⼈，既要技术达标⼜要和已招的⼈技能互补（迭代过程）。了解下使⽤最⼤边际相关性的场景，更能让我们理解其概念：• 语义相似性使⽤场景：搜索引擎的基础排序、重复检测、聚类、语义搜索。• MMR 使⽤场景：推荐系统：推荐与⽤⼾兴趣相关但⼜不同类型的物品，避免“信息茧房”。⽂档摘要：从⻓⽂档中选择能代表主旨⼜包含不同信息的句⼦，避免摘要内容重复。RAG (检索增强⽣成)：在从知识库检索完⼀堆相关⽂档后，使⽤ MMR 进⾏去重和多样化筛选，再交给LLM⽣成答案，能有效提升答案质量和减少幻觉。通过 ngram 重叠选择⽰例（Ngram）
什么是【ngram】？ngram 指⼀个⽂本序列中连续的 n 个词（word） 或字符（character）。什么是【ngram 重叠】？通过计算它们之间共同拥有的 ngram 数量来⼀种衡量两段⽂本相似度的⽅法。例如下述两段⽂本：text1 = "苹果⼿机很好⽤" (分词后: 苹果 ⼿机 很 好⽤ )text2 = "这款⼿机很好⽤" (分词后: 这款 ⼿机 很 好⽤ )这两段⽂本单词重复度很⾼，连续三个词的相同的情况也存在，因此 ngram 重叠⾼。再看个例⼦：text1 = 苹果⼿机很好⽤" (分词后: 苹果 ⼿机 很 好⽤ )text2 = "iPhone ⾮常不错" (分词后: iPhone ⾮常 不错 )这两段⽂本在含义上⾮常相似，但它们的 ngram 重叠度为 0。因此，传统 ngram 重叠是⼀种表⾯形式的匹配。它只关⼼词是否完全⼀样，但对于同义词却⽆法处理。什么是【语义 ngram 重叠】？不再⽐较词本⾝，⽽是⽐较词背后的语义向量（Embedding）。也就是说，它不是看两个词 [ 苹果 ] 和 [iPhone] 的字⾯是否相同，⽽是计算它们在语义空间中的向量是否相似。如果相似度超过某个阈值，就认为它们“重叠”了。还是看这个例⼦：text1 = 苹果⼿机很好⽤" (分词后: 苹果 ⼿机 很 好⽤ )text2 = "iPhone ⾮常不错" (分词后: iPhone ⾮常 不错 )•计算 苹果 和 iPhone 的向量相似度 → 得分 0.95 (很⾼，视为重叠)•计算 ⼿机 和 iPhone 的向量相似度 → 得分 0.88 (很⾼，但可能不会同时计分，取决于算法设计，避免重复计算)•计算 很 和 ⾮常 的向量相似度 → 得分 0.90 (很⾼，视为重叠)•计算 好⽤ 和 不错 的向量相似度 → 得分 0.82 (很⾼，视为重叠)•最终，语义上的 unigram 重叠度可能为 3 或 4（⾮常相似！）。那么语义 ngram 重叠的使⽤场景是什么？语义 ngram 重叠常⽤于需要更精准语义评估的场景，例如 剽窃检测 ， 能够发现那些改换了词汇但保留了核⼼思想的“智能”剽窃。from langchain_community.example_selectors import NGramOverlapExampleSelector from langchain_core.prompts import FewShotPromptTemplate, PromptTemplate # 翻译⽰例 examples = [ {"input": "See Spot run.", "output": "看⻅Spot跑。"}, {"input": "My dog barks.", "output": "我的狗叫。"}, {"input": "Spot can run.", "output": "Spot可以跑。"}, ] # 字符串模板 example_prompt = PromptTemplate( input_variables=["input", "output"], template="Input: {input}\nOutput: {output}", ) # NGram ⽰例选择器 example_selector = NGramOverlapExampleSelector( examples=examples, example_prompt=example_prompt, threshold=-1.0, ) # ⽤于实例化⽰例的模板 dynamic_prompt = FewShotPromptTemplate( example_selector=example_selector, example_prompt=example_prompt, prefix="给出每个输⼊的中⽂翻译", suffix="Input: {sentence}\nOutput:", input_variables=["sentence"], ) # 按照重叠分数排序 print( dynamic_prompt.invoke({"sentence": "Spot can run fast."}).to_messages() [0].content )结果如下
给出每个输⼊的中⽂翻译 Input: Spot can run. Output: Spot可以跑。 Input: See Spot run. Output: 看⻅Spot跑。 Input: My dog barks. Output: 我的狗叫。 Input: Spot can run fast. Output:可以看到结果集是按照 ngram 重叠分数进⾏了排序。可以分别设置 threshold=0.0 和 1.0 进⾏测试，打印结果如下：threshold=0.0 （表⽰排除不相关的⽰例） 给出每个输⼊的中⽂翻译 Input: Spot can run. Output: Spot可以跑。 Input: See Spot run. Output: 看⻅Spot跑。 Input: Spot can run fast. Output: threshold=1.0（排除所有⽰例） 给出每个输⼊的中⽂翻译 Input: Spot can run fast. Output：
输出解析器（Output parsers）
负责获取模型的输出，并将输出转换为更结构化的格式。当使⽤ LLM ⽣成结构化数据或规范化聊天模型和 LLM 的输出时，这很有⽤。⼤型语⾔模型（LLM）的输出本质上是⾮结构化的⽂本。但在构建应⽤程序时，我们通常希望得到结构化的、机器可读的数据，这样可以将其转换为更适合下游任务的格式，⽐如：• JSON 对象• Python 字典或列表• ⼀个特定的 Pydantic 模型实例• ⼀个简单的布尔值或字符串枚举输出解析器的作⽤就是架起这座桥梁：它们将 LLM 的⾮结构化⽂本输出转换为结构化格式。这使得与LLM 的交互从“模糊的⽂本对话”变成了“精确的数据 API 调⽤”，是构建可靠、⾼效 LLM 应⽤不可或缺的组件。解析⽂本输出
其实对于使⽤ StrOutputParser 输出解析器输出⽂本，我们已经使⽤过多次了。对于StrOutputParser ，它也实现了标准的 Runnable 接⼝。⽰例如下：from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser model = ChatOpenAI(model="gpt-4o-mini") chain = model | StrOutputParser() for chunk in chain.stream("写⼀⾸夏天的诗词，50字以内。"): print(chunk, end="|")结果如下
|炎|夏|骄|阳|照|，|绿|树|映|蓝|天|。| |蝉|鸣|声|声|烈|，|荷|塘|映|清|鲜|。| |微|⻛|拂|⾯|过|，|凉|意|透|⼼|间|。| |烦|忧|随|汗|去|，|畅|享|此|夏|欢|。||解析结构化对象输出
要输出结构化对象，需要⽤到的输出解析器是 PydanticOutputParser 。class langchain_core.output_parsers.pydantic. PydanticOutputParser 类，其参数如下：• pydantic_object ：要解析的 pydantic 模型。内置⽅法：invoke() ：将单个输⼊转换为输出。get_format_instructions() → str ：重要！！• 作⽤：⽣成⼀个指令字符串，这个字符串会被添加到发送给 LLM 的提⽰（Prompt）的末尾。• ⽬的：告诉 LLM 应该以什么样的格式返回它的响应。例如，“请将你的回复封装在 XML 标签中”或“请以 JSON 格式输出，包含 ‘name’ 和 ‘age’ 两个字段”。from langchain_openai import ChatOpenAI from langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import PromptTemplate from typing import Optional from pydantic import BaseModel, Field # 定义⼤模型 model = ChatOpenAI(model="gpt-4o-mini") # 定义输出结构：Pydantic 类 class Joke(BaseModel): """给⽤⼾讲⼀个笑话。""" setup: str = Field(description="这个笑话的开头") punchline: str = Field(description="这个笑话的妙语") rating: Optional[int] = Field( default=None, description="从1到10分，给这个笑话评分" ) # 设置解析器 parser = PydanticOutputParser(pydantic_object=Joke) # 提⽰词模板 prompt = PromptTemplate( template="Answer the user query.\n{format_instructions}\n{query}\n", input_variables=["query"], # partial_variables：提⽰模板携带的部分变量的字典，⽆需在每次调⽤提⽰时都传⼊它们。 # 类型为 Mapping[str, Any]，传⼊template携带的部分变量的字典。 partial_variables={"format_instructions": parser.get_format_instructions()}, ) chain = prompt | model | parser for chunk in chain.stream({"query": "给我讲⼀个关于唱歌的笑话"}): print(chunk, end="|")结果如下setup='为什么歌⼿总是带着铅笔去演出？' punchline='' rating=None| setup='为什么歌⼿总是带着铅笔去演出？' punchline='因为' rating=None| setup='为什么歌⼿总是带着铅笔去演出？' punchline='因为他们' rating=None| setup='为什么歌⼿总是带着铅笔去演出？' punchline='因为他们想' rating=None| setup='为什么歌⼿总是带着铅笔去演出？' punchline='因为他们想要' rating=None| setup='为什么歌⼿总是带着铅笔去演出？' punchline='因为他们想要不断' rating=None| setup='为什么歌⼿总是带着铅笔去演出？' punchline='因为他们想要不断调整' rating=None| setup='为什么歌⼿总是带着铅笔去演出？' punchline='因为他们想要不断调整⾳' rating=None| setup='为什么歌⼿总是带着铅笔去演出？' punchline='因为他们想要不断调整⾳调' rating=None| setup='为什么歌⼿总是带着铅笔去演出？' punchline='因为他们想要不断调整⾳调！' rating=None|解析 JSON 输出
要输出 JSON 格式，需要⽤到的输出解析器是 JsonOutputParser 。class langchain_core.output_parsers.json. JsonOutputParser 类，其参数如下：• pydantic_object ：⽤于验证的 Pydantic 对象。如果为空，则不执⾏任何验证。内置⽅法：invoke() ：将单个输⼊转换为输出。get_format_instructions() → str ：重要！！• 作⽤：⽣成⼀个指令字符串，这个字符串会被添加到发送给 LLM 的提⽰（Prompt）的末尾。• ⽬的：告诉 LLM 应该以什么样的格式返回它的响应。例如，“请将你的回复封装在 XML 标签中”或“请以 JSON 格式输出，包含 ‘name’ 和 ‘age’ 两个字段”。
文档加载器（Document loaders）
RAG
RAG 概念
RAG 阶段（Retrieval-Augmented Generation，检索增强⽣成）。这是当前⼤语⾔模型应⽤的核⼼模式。RAG 的流程相对复杂，为了更好的理解 RAG，我们先⽤ AI 搜索来引出 RAG。对于【AI ⼤模型】来说，它最擅⻓的是语义理解和⽂本总结，最不擅⻓的就是获取实时的信息。因为⼤模型的训练数据是有截⽌⽇期的！对于【搜索引擎】来说，它最擅⻓的就是获取实时的信息，缺点是信息分散，每次都需要⼈为进⾏总结。⼤模型与搜索引擎的结合，就是给 AI 配备了⼀个活字典，让 AI 可以随时进⾏查阅。
有了以上流程的铺垫，接下来，正式进⼊ RAG 的学习。首先，先来思考⼀个问题：搜索引擎可以帮我们解决实时数据的获取，但获取到的数据也是受限的。它只能获取到公开在⽹络中的数据，而无法获取到⼀些本地数据，或企业内部的私有数据等，此时该如何？答案是使⽤ RAG（检索增强⽣成）技术！当用户向 LLM 提问时，系统⾸先在知识库（如公司内部⽂档）中进⾏语义搜索，找到最相关的内容，然后将这些内容和问题⼀起交给 LLM 来⽣成答案。与 AI 搜索类⽐，本质是知识库改变了，从搜索引擎线上搜索改为了本地或私有知识库中搜索。RAG 流程
RAG 的流程分为【离线数据处理】和【在线检索】两个过程。上⾯提到，RAG 知识库可以是本地⽂档、公司内部⽂档等⼀些私有化数据。但这些私有数据或⽂档实际上并不能很好地被直接进⾏检索访问。因此需要将这些私有化数据构建成可以被检索的知识库，这就是离线数据处理要⼲的事情。经过离线数据后，知识则会按照某种格式以及排列⽅式存储在知识库中，等待被使⽤。⽽在线检索则是我们依赖知识库查询，通过⼤模型⽣成结果的过程。过程如下图所⽰：
这张图将会是我们后续要学习 LangChain 组件知识地图，所有的组件都会⼀ 进⾏讲解，现在我们只需掌握其流程，接触相关概念即可。• ⽂档加载 (Document Loading)：加载多种不同来源加载⽂档。LangChain 提供了 100 多种不同的⽂档加载器，包括 PDF 在内的⾮结构化的数据、SQL 在内的结构化的数据，以及 Python、Java之类的代码等。• ⽂本分割 (Splitting)：⽂本分割器把 Documents 切分为指定⼤⼩的块。• 存储 (Storage)：存储涉及到两个环节，分别是：将切分好的⽂档块进⾏嵌⼊（Embedding），即将⽂档块转换成向量的形式。将 Embedding 后的向量数据，存储到向量数据库中。• 检索 (Retrieval)：数据存⼊向量数据库后。当我们需要进⾏数据检索时，会通过某种检索算法找到与输⼊问题相似的⽂档块。• 输出 (Output)：把问题以及检索出来的⽂档块⼀起提交给 LLM，LLM 会通过问题和检索出来的提⽰⼀起来⽣成更加合理的答案。我们现在已经知道了 RAG 的完整流程，但也仅是知道 RAG 是什么，⾄于流程中为什么要进⾏⽂档加载、⽂本分割、存储，我们还⽆从了解。因此后续的各个 LangChain 组件讲解时都会涉及每个步骤被设计出来的原因。RAG ⽰例
讲这么多，不如⼀⻅。接下来就来演⽰⼀个 RAG 系统。在这个⽰例中，我们• 提供了：《租房项⽬Q&A》⽂档• 希望：通过聊天⽅式，提问关于项⽬的任何问题，最终得到答案。• 要求：最多只⽤三句话回答，要简明扼要。代码如下（不⽤理解代码含义，只需要看结果即可）：from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_redis import RedisConfig, RedisVectorStore from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough # 定义聊天模型 model = ChatOpenAI(model="gpt-4o-mini") # 定义嵌⼊模型 embeddings = OpenAIEmbeddings(model="text-embedding-3-large") # 配置 Redis 客⼾端 redis_url = "redis://192.168.100.238:6379" config = RedisConfig( index_name="qa", redis_url=redis_url, metadata_schema=[ {"name": "category", "type": "tag"}, {"name": "num", "type": "numeric"}, ], ) # 定义 Redis 向量存储 vector_store = RedisVectorStore(embeddings, config=config) # ⽣成检索器 retriever = vector_store.as_retriever() # 定义提⽰词模板 prompt = ChatPromptTemplate.from_messages( [ ( "human", """你是负责回答问题的助⼿。使⽤以下检索到的上下⽂⽚段来回答问题。如果你不知 道答案，就说你不知道。最多只⽤三句话，回答要简明扼要。 Question: {question} Context: {context} Answer:""", ), ] ) # 将⽂档转换为字符串 def format_docs(docs): return "\n\n".join(doc.page_content for doc in docs) # 定义链 rag_chain = ( {"context": retriever | format_docs, "question": RunnablePassthrough()} | prompt | model | StrOutputParser() ) # 循环输⼊问题 while True: # 获取⽤⼾输⼊ question = input("\n请输⼊您的问题（输⼊'退出'或'quit'结束程序）: ").strip() # 检查是否退出 if question.lower() in ["退出", "quit"]: print("程序已结束，再⻅！") break # 检查输⼊是否为空 if not question: print("问题不能为空，请重新输⼊。") continue # 执⾏链，流式输出 print("回答: ", end="", flush=True) chunks = [] for chunk in rag_chain.stream(question): chunks.append(chunk) print(chunk, end="", flush=True) print() # 换⾏请输⼊您的问题（输⼊'退出'或'quit'结束程序）: 介绍⼀下这个项⽬ 回答: 这个项⽬是⼀个基于脚⼿架的微服务在线租房系统，旨在模仿⻉壳、安居客等流⾏应⽤，具备真 实的交互体验和架构设计。通过结合理论和实践，我在这个项⽬中加深了对 Java 编程语⾔的理解。这 个项⽬的灵感来源于我在⼤学时和同学合租的经历。 请输⼊您的问题（输⼊'退出'或'quit'结束程序）: 项⽬设计难点有哪些？ 回答: 项⽬设计的难点主要包括⾮技术⽅⾯的需求把控和技术⽅⾯的明确项⽬范围与⽬标。缺乏产品经 理的介⼊使得设计过程中需⾃⾏绘制原型图并拆分功能，这增加了设计的复杂性。同时，测试和质量保 证也是项⽬中的重要挑战，需要合理安排时间以确保项⽬质量。 请输⼊您的问题（输⼊'退出'或'quit'结束程序）: 介绍数据存储的相关设计 回答: 数据存储的设计中，使⽤MySQL作为关系数据库管理系统，并结合MyBatis简化SQL操作。为了 更好地应对⾼并发场景，设计了Redis缓存⽅案来优化内存利⽤和减轻数据库压⼒。引⼊OSS对象存储 是为了实现⽆限扩展性，相对于本地存储和MySQL，它具备更⾼的性能和成本效益。 请输⼊您的问题（输⼊'退出'或'quit'结束程序）: 详细介绍 Redis 与 Mysql 数据⼀致性⽅案 回答: Redis与MySQL的数据⼀致性⽅案通常采⽤“双写⼀致性”模式，通过Cache-Aside⽅法实现。在 这⼀⽅案中，应⽤程序在读取数据时先查询Redis缓存，如果不存在再查询MySQL数据库，并将结果缓 存到Redis中；在更新数据时，需同时更新MySQL和Redis，以确保两者状态⼀致。此⽅法也包括设置 合理的缓存过期时间和使⽤布隆过滤器，以解决缓存穿透等问题。 请输⼊您的问题（输⼊'退出'或'quit'结束程序）:我们提供的文档越详细，⽣成的结果越符合预期。Document ⽂档类
要想实现 RAG，⾸先就需要从源中获取数据，即加载数据或⽂档。这是通过 LangChain 的⽂档加载器完成的。LangChain ⽂档加载器可以将各种数据源加载成⼀系列的⽂档对象 Document 。class langchain_core.documents.base.Document ⽤于存储⼀段⽂本和相关元数据的类，我们可以直接定义LangChain ⽂档列表，如下所⽰from langchain_core.documents import Document documents = [ # 单个Document对象通常表⽰较⼤⽂档的⼀个块 Document( # 内容字符串 page_content="狗是很好的伴侣，以忠诚和友好⽽闻名。", # 元数据字典 # 元数据属性可以捕获有关⽂档源、与其他⽂档的关系以及其他信息的信息。 metadata={"source": "mammal-pets-doc"}, ), Document( page_content="猫是独⽴的宠物，经常享受⾃⼰的空间。", metadata={"source": "mammal-pets-doc"}, ), ]这⾥我们定义了⼀个 documents ⽂档列表，其内包含了两个 Document ⽂档对象。通常，单个Document对象表⽰较⼤⽂档的⼀个块/⻚。每个 Document 对象，包含了以下参数：• id ：可选的⽂档标识符。理想情况下，这应该在整个⽂档集合中是唯⼀的，并格式化为UUID，但不会强制执⾏。• page_content ：字符串⽂本• metadata ：与内容关联的任意元数据。类型为 dict [Optional]加载 PDF ⽂档
将本地的 PDF ⽂档加载到 LangChain 中，其实就是将 PDF ⽂档转换为⼀个个 Document 对象。这时就需要我们使⽤ PyPDFLoader ⽂档加载器完成这⼀功能。class langchain_community.document_loaders.pdf. PyPDFLoader 类，有以下关键函数：• init() 初始化函数，⼊参 file_path ，表⽰要加载的 PDF ⽂件的路径。• load() → list[Document] ：将数据加载到⽂档对象中。返回⽂档对象列表。现在，让我们加载⼀个本地 PDF ⽂档看下效果。from langchain_community.document_loaders import PyPDFLoader file_path = "../Docs/PDF/脚⼿架级微服务租房平台Q&A.pdf" loader = PyPDFLoader(file_path) # 将 PDF ⽂件的每⼀⻚转换为⼀个独⽴的 Document 对象，并存储在列表 docs 中。 docs = loader.load()print(f"问：PDF ⽂件的总⻚数为：\n{len(docs)}\n") print(f"问：第⼀⻚⽂本内容的前200个字符是：\n{docs[0].page_content[:200]}\n") print(f"问：第⼀⻚元数据：\n{docs[0].metadata}")结果如下
问：PDF ⽂件的总⻚数为： 32 问：第⼀⻚⽂本内容的前200个字符是： 脚⼿架级微服务租房平台 通⽤问题 1. 为什么做这个项⽬？ • 回答1：（出于兴趣爱好开发） ⼤学期间，我和同学在外合租过⼀段时间，使⽤了⼀些租房平台，于是我有个想法，⾃⼰能不能开发 ⼀个租房平台，可以让我将理论知识与实践相结合。我希望通过实际项⽬来加深对Java编程语⾔和相 关技术的理解。于是我便查找了⼀些资料，看了⼀些开源项⽬，进⾏了⼀些改进。 • 回答2：（开源项⽬的解释） 这个 问：第⼀⻚元数据： {'producer': 'pdfcpu v0.8.1 dev', 'creator': 'Chromium', 'creationdate': '2025-08-28T17:52:34+08:00', 'moddate': '2025-08-28T17:52:34+08:00', 'source': '../PDF/脚⼿架级微服务租房平台Q&A.pdf', 'total_pages': 32, 'page': 0, 'page_label': '1'}print(f"问：PDF ⽂件的总⻚数为：\n{len(docs)}\n") print(f"问：第⼀⻚⽂本内容的前200个字符是：\n{docs[0].page_content[:200]}\n") print(f"问：第⼀⻚元数据：\n{docs[0].metadata}")问：PDF ⽂件的总⻚数为： 32 问：第⼀⻚⽂本内容的前200个字符是： 脚⼿架级微服务租房平台 通⽤问题 1. 为什么做这个项⽬？ • 回答1：（出于兴趣爱好开发） ⼤学期间，我和同学在外合租过⼀段时间，使⽤了⼀些租房平台，于是我有个想法，⾃⼰能不能开发 ⼀个租房平台，可以让我将理论知识与实践相结合。我希望通过实际项⽬来加深对Java编程语⾔和相 关技术的理解。于是我便查找了⼀些资料，看了⼀些开源项⽬，进⾏了⼀些改进。 • 回答2：（开源项⽬的解释） 这个 问：第⼀⻚元数据： {'producer': 'pdfcpu v0.8.1 dev', 'creator': 'Chromium', 'creationdate': '2025-08-28T17:52:34+08:00', 'moddate': '2025-08-28T17:52:34+08:00', 'source': '../PDF/脚⼿架级微服务租房平台Q&A.pdf', 'total_pages': 32, 'page': 0, 'page_label': '1'}现在许多 LLM ⽀持对多模态输⼊（例如图像）进⾏推理。在某些应⽤程序中，例如对具有复杂布局、图表或扫描的 PDF 进⾏问答，可以跳过 PDF 解析，直接将 PDF ⻚⾯转换为图像并将其直接传递给模型可能是更准确的加载 Markdown ⽂件
就像用一台会分层扫描的智能扫描仪，把 .md 文档转化成工坊能处理的“配方卡片”（Document 对象）。
用什么工具？——
UnstructuredMarkdownLoader它负责把 Markdown 文件读进来。但有个前提：要先安装解析引擎
两种扫描模式：
single和elements你可以告诉扫描仪用哪种精细度工作，这由
mode参数决定。1.
mode="single"：整本扫描，只出一张卡片干什么：把整个 .md 文件当成一个整体，装进唯一一个 Document 对象里。
比喻：就像把一整本配方手册直接复印成一张大卡片，不分章节。
结果：
docs[0].page_content包含了全文；docs[0].metadata里只有简单的来源信息（比如文件路径）。适用场景：文件很小，或者你不需要精确到段落去检索。
2.
mode="elements"：分层扫描，逐块出卡片干什么：智能识别 Markdown 里的标题、段落、列表、表格、图片等元素，每种元素单独生成一个 Document。
比喻：像档案员把手册拆成一颗颗“积木块”，每块贴上标签（类型、层级、所属关系）。
结果：可能生成长达几百个细碎的 Document。
每个 Document 的元数据（
metadata） 里包含关键字段：
category：这块是什么类型（Title、NarrativeText、ListItem、Table等）
element_id：这块自身的唯一 ID
parent_id：它从属于哪一块（比如某个段落的parent_id指向它所属的二级标题）通过
parent_id，你能把拆散的块重新拼回完整的文档层级——就像乐高积木自带组装说明书。
适用场景：需要精细检索（比如只搜“回答”部分，不搜“问题”部分），或者需要理解文档结构时。
为什么要这么分？
因为大模型的工作台（上下文窗口）大小有限，切细了能塞进更多相关片段；同时保留结构信息后，你在检索时还能根据层级做更精准的过滤，比如“只从所有二级标题下的段落里找答案”。
对于 LangChain 来说，能加载的⽂档类型远不⽌这些，它还能加载⽹⻚、⼀些云提供商⽂件、社交媒 体平台⽂档等
文本分割器（Text splitters）
我们已经知道可以通过⽂档加载器完成各种数据源的加载，将其转换为⽂档对象 Document 。那么接下来要做的就是⽂档拆分。⽂档拆分通常是将⼤⽂本分解为更⼩的、易于管理的块。这对于索引数据并将其传递到模型中都很有⽤。因为，⼤块更难搜索并且不适合模型的有限上下⽂窗⼝。拆分可以提⾼搜索结果的粒度，从⽽可以更精确地将查询与相关⽂档部分进⾏匹配。LangChain 的⽂本分割器便能将⼤型⽂档分解为更⼩的块。根据⽂档⻓度与⽂档语义拆分
我们可以直接根据⽂档的⻓度拆分⽂档，是最简单且有效的⽅法。可确保每个块不超过指定的⼤⼩限制。对于⻓度拆分，其实也分为两种： 基于字符⻓度拆分 和 基于Token ⻓度拆分 。基于字符⻓度拆分
基于 Token ⻓度拆分
硬性约束⻓度拆分
特殊⽂档结构拆分
若对于代码等特殊⽂本，可以尝试使⽤ Language 提供的不同的分割器（如PythonCodeTextSplitter 、 HTMLHeaderTextSplitter 等）效果会更好，它会理解代码的语法结构。这⾥了解下常⻅的拆分原则即可：• Markdown：根据标头拆分（例如，#、##、###）• HTML：使⽤标签拆分• JSON： 按对象或数组元素拆分• Code 代码 ：按函数、类或逻辑块拆分
文本向量
嵌⼊与嵌⼊模型（Embedding and Embedding Models）计算机天⽣擅⻓处理数字，但不理解⽂字、图⽚的含义。嵌⼊（Embedding）的核⼼思想就是将⼈类世界的符号（如单词、句⼦、产品、⽤⼾、图⽚）转换为计算机能够理解的数值形式（即向量，本质上是⼀个数字列表），并且要求这种转换能够保留原始符号的语义和关系。我们可以把它想象成⼀个翻译过程，把⼈类语⾔“翻译”成计算机的“数学语⾔”说明：我们之前⼀直⽤的⼤语⾔模型是⽣成式模型。它理解输⼊并⽣成新的⽂本（回答问题、写⽂章）。它内部实际上也使⽤嵌⼊技术来理解输⼊，但最终⽬标是“创造”。⽽嵌⼊模型（Embedding Models）是表⽰型模型。它的⽬标不是⽣成⽂本，⽽是为输⼊的⽂本创建⼀个最佳的、富含语义的数值表⽰（向量）。如 OpenAI 的 "text-embedding-3-large" 嵌⼊模型；Google 的 "gemini-embedding-001" 嵌⼊模型；阿⾥的 "Qwen3-Embedding-8B" 嵌⼊模型等。
在当今的大模型应用开发（尤其是 RAG 检索增强生成）中，**文本向量**与**向量数据库**已经是核心技术基石。传统的关键词搜索无法理解语义，而向量搜索能真正“读懂”文字背后的含义。本文将基于实战开发的章节结构，带你从理论到代码，彻底掌握嵌入模型、不同维度的向量存储、以及元数据过滤和 MMR 搜索等高级操作。什么是文本向量与嵌入模型？
什么是向量？
计算机无法直接理解“苹果”和“香蕉”是否相似。嵌入模型（Embedding Model）会将文本转化为一组多维浮点数（即**向量**）。在数学空间中，语义相似的文本，其向量距离会非常接近。比如“天气真好”和“阳光明媚”的向量余弦相似度就会很高。嵌入模型的应用场景
语义搜索：直接匹配用户意图，而非仅仅匹配关键词。
大模型 RAG：将私有知识库转换为向量，让大模型在回答时提供事实依据。
推荐系统：根据用户喜好向量，推荐相似的文章或商品。在代码中定义与使用 Embeddings 模型
在具体开发中，我们通常使用 `LangChain` 等框架来调用嵌入模型。
定义嵌入模型
最常用的是 OpenAI 的 `text-embedding-3-small` 或 Hugging Face 开源模型（如 `BAAI/bge-large-zh`）。嵌入文档列表
假设我们有一堆文档（如 PDF 或 TXT 的段落），我们需要批量将其转化为向量并存储在数据库中。这一步决定了后续搜索的质量。嵌入单个查询
当用户提问时（比如“如何开通海外账户？”），需要将这句话实时转化为向量，然后去数据库里进行匹配搜索。核心组件：向量存储与向量数据库
嵌入后的向量不能放在普通关系型数据库（如 MySQL）中，因为普通 SQL 无法进行高效的向量相似性计算。
向量数据库介绍
它是一个专门为存储向量和进行**最近邻搜索（ANN）**而设计的数据库。例如：Redis, Pinecone, Qdrant, Milvus 等。轻量级测试：内存存储
在写 Demo 或做单元测试时，我们不需要部署真正的数据库。可以直接在内存中存储向量
基本操作：创建存储实例，添加文档（ID + 向量 + 元数据）。
向量搜索：直接在内存中计算向量距离，返回 Top-K 结果。它速度快但不支持持久化。真实落地：Redis 向量存储
Redis 是持久化且支持索引的高性能数据库。
基本概念与环境设置
需要安装 `redis-stack` 版本（自带向量搜索插件），安装后配置 Redis URL。基本操作
创建 Redis 实例，定义索引模式（比如定义向量维度为 1536，距离算法为余弦相似度），然后批量插入向量数据。向量搜索：不仅仅是相似性
这是 Redis 搜索的精华所在：相似性搜索：最基础的模式，比如“帮我在回忆录里找关于童年的文章”，返回语义最相似的片段。
元数据过滤：这是一项**实战必杀技**。比如用户要搜“2010年以前关于电影的所有评论”。我们不能只搜向量，还需要过滤年份 `year < 2010`。Redis 可以完美结合向量搜索和结构化数据过滤。
最大边际相关性搜索（MMR）：常用于 RAG 场景。如果直接搜索 Top-5 相似文档，可能会返回 5 个内容高度重复的段落。MMR 可以在保证相关性的前提下，增加结果的多样性，避免大模型被冗余信息干扰。企业级首选：Pinecone 向量存储
如果不想自己部署运维 Redis 或 Milvus，Pinecone 是目前托管向量数据库的行业标杆。
Pinecone 介绍
全托管服务，免运维，支持高达 99.99% 的可用性，非常适合快速迭代的产品。环境设置与基本操作
注册 Pinecone 账号 -> 获取 API Key -> 创建 Index（指定维度和度量标准）。Pinecone 的操作逻辑和 Redis 类似，但只需要几行代码就能连接云端。向量搜索进阶
在 Pinecone 中，你可以非常轻松地组合 **相似性搜索** 与 **元数据过滤**。比如在电商场景中，搜索“红色男款鞋子”并过滤掉库存为 0 的商品。
检索器（Retrievers）
检索系统
检索系统（Information Retrieval System, IR System）是⼀个为了满⾜⽤⼾信息需求，从⼤规模、⾮结构化的数据集合中，⾃动、⾼效地查找、排序并返回相关信息的计算机系统。它的核⼼任务是：在正确的时间，以正确的⽅式，将正确的信息传递给正确的⼈。最常⻅的例⼦就是：搜索引擎（如Google、百度）。随着⼤型语⾔模型的流⾏，检索系统已成为⼈⼯智能应⽤（例如 RAG）的重要组成部分。且存在多种【不同类型】的检索系统，包括：• 关系数据库关系数据库是许多应⽤程序中使⽤的结构化数据存储的基本类型。数据存储在⾏（记录）和列（属性）中，可以通过 SQL（结构化查询语⾔）进⾏⾼效的查询和作。关系数据库擅⻓维护数据完整性、⽀持复杂查询以及处理不同数据实体之间的关系。• 词法搜索索引许多搜索引擎基于将查询中的单词与每个⽂档中的单词进⾏匹配，这种⽅法称为词法检索。即⼀个单词经常出现在⽤⼾的查询和特定⽂档中，那么这个⽂档可能是⼀个很好的匹配。这通常使⽤【倒排索引】实现。• 向量数据库向量存储不使⽤字频，⽽是使⽤【嵌⼊模型】将⽂档转换为⾼维向量表⽰。这允许使⽤余弦相似度等数学运算对嵌⼊向量进⾏有效的相似性搜索。检索器
检索器是检索系统中的⼀个核⼼组件，它接收来⾃⽤⼾接⼝的查询（Query），检索出包含查询关键词的候选⽂档集合。我们可以使⽤上⾯提到的任何检索系统实现⽅式创建检索器！如关系数据库、向量数据库等。由于其重要性和多样性，LangChain 提供了⼀个统⼀的接⼝来与不同类型的检索系统进⾏交互。LangChain的检索器接⼝⾮常简单：1. 输⼊：查询字符串2. 输出：⽂档列表（标准化的 LangChain ⽂档对象 Document）使⽤向量数据库作为检索器
向量存储是索引和检索⾮结构化数据的⼀种强⼤⽽有效的⽅法。可以通过调⽤向量数据库的as_retriever ⽅法，将向量存储⽤作检索器。在这⾥我们使⽤ Redis 向量存储。from langchain_openai import OpenAIEmbeddings from langchain_redis import RedisConfig, RedisVectorStore # 定义嵌⼊模型 embeddings = OpenAIEmbeddings(model="text-embedding-3-large") # 配置 Redis 客⼾端 redis_url = "redis://192.168.100.238:6379" config = RedisConfig( index_name="qa", redis_url=redis_url, metadata_schema=[ {"name": "category", "type": "tag"}, {"name": "num", "type": "numeric"}, ], ) # Redis 存储初始化 vector_store = RedisVectorStore(embeddings, config=config) retriever = vector_store.as_retriever() docs = retriever.invoke("数据库表怎么设计的？") for doc in docs: print("*" * 30) print(doc.page_content[:30])LangChain 检索器是⼀个 Runnable 的对象 ，它是 LangChain 组件的标准接⼝。这意味着它有⼀些常⽤⽅法，包括 invoke ，⽤于与其交互。默认情况下，向量存储检索器使⽤相似性搜索。 输出结果如下：****************************** 提供两种⽅案有以下好处： 降低⻔槛：单机版让⽤⼾快速体验核 ****************************** 数据库：MySQL + MyBatis MySQL 和 M ****************************** 不同状态（上架/已租/下架）的查询频率不同，拆分可减少锁争⽤ ****************************** 分库分表：按user_id哈希分库，按⽉份分表（如chat_使⽤ @chain 创建“检索器”
除了使⽤ as_retriever ⽅法，我们还可以⾃⾏创建⼀个“检索器”。回想⼀下检索器的特点：1. LangChain 检索器是⼀个 Runnable 的对象2. LangChain 检索器输⼊为查询字符串，输出为⽂档列表（标准化的 LangChain ⽂档对象Document）综上所述，我们可以：from langchain_core.runnables import chain from typing import List from langchain_core.documents import Document @chain def retriever(query: str) -> List[Document]: return vector_store.similarity_search(query, k=2) docs = retriever.invoke("数据库表怎么设计的？") for doc in docs: print("*" * 30) print(doc.page_content[:30])上⾯定义了⼀个函数，使⽤ @chain 修饰，该修饰可以使其成为 Runnable 函数，且满⾜检索器输⼊输出的要求。在函数中，我们依旧使⽤向量数据库的相似性搜索⽅法，这样灵活性也更⾼，想要进⾏元数据筛选也更⽅便。注意，这并不是真正的检索器，检索器是⼀个 Runnable 对象，⽽我们定义的只是⼀个函数，具备其特点罢了。
RAG 案例
RAG 是当前⼤语⾔模型应⽤的核⼼模式。当⽤⼾向 LLM 提问时，系统⾸先使⽤嵌⼊模型在知识库中进⾏语义搜索，找到最相关的内容，然后将这些内容和问题⼀起交给 LLM 来⽣成答案。这极⼤地提⾼了答案的准确性和时效性。由于 LangChain 检索器是⼀个 Runnable 的对象，我们便可以⽅便的使⽤ 链 完成相关的调⽤。下⾯我们来完成⼀个最简单的 RAG 案例，将会完成：1. 根据 Query 搜索最相关的 4 篇⽂档。2. 将相关的⽂档转换为字符串，以便后续发送给聊天模型。3. 将 Query 与⽂档字符串发送给聊天模型。4. 聊天模型依据输出解析器格式输出内容。from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_redis import RedisConfig, RedisVectorStore from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough # 定义聊天模型 model = ChatOpenAI(model="gpt-4o-mini") # 定义嵌⼊模型 embeddings = OpenAIEmbeddings(model="text-embedding-3-large") # 配置 Redis 客⼾端 redis_url = "redis://192.168.100.238:6379" config = RedisConfig( index_name="qa", redis_url=redis_url, metadata_schema=[ {"name": "category", "type": "tag"}, {"name": "num", "type": "numeric"}, ], ) # 定义 Redis 向量存储 vector_store = RedisVectorStore(embeddings, config=config) # ⽣成检索器 retriever = vector_store.as_retriever() # 定义提⽰词模板 prompt = ChatPromptTemplate.from_messages( [ ( "human", """你是负责回答问题的助⼿。使⽤以下检索到的上下⽂⽚段来回答问题。如果你不知 道答案，就说你不知道。最多只⽤三句话，回答要简明扼要。 Question: {question} Context: {context} Answer:""", ), ] ) # 将⽂档转换为字符串 def format_docs(docs): return "\n\n".join(doc.page_content for doc in docs) # 定义链 rag_chain = ( {"context": retriever | format_docs, "question": RunnablePassthrough()} | prompt | model | StrOutputParser() ) # 执⾏链，流式输出 for chunk in rag_chain.stream("数据库表怎么设计的？"): print(chunk, end="|", flush=True)上述代码中， RunnablePassthrough 我们之前还没有⻅过，简单来说，RunnablePassthrough 是⼀个“伪”Runnable，它的主要作⽤是在链（Chain）中透明地传递输⼊数据，⽽不做任何修改。当我们需要将原始输⼊和另⼀个处理过程的输出⼀起传递给下⼀个步骤时，就需要RunnablePassthrough 。就例如代码中，我们需要将【Query】与【通过检索出来的⽂档转换的字符串】同时发送给提⽰词模板。打印结果如下|数据库|表|设计|可以|通过|⽔平|分|表|来|优化|性能|，|特别|是|当|单|表|数据|量|过|⼤| 时|，可以|减少|锁|争|⽤|和|提升|查询|效率|。|实现|⽅式|是|将|状态|字段|独|⽴|存|储| 在|house|_status|表|中|，并|通过|house|_id|进⾏|关联|查询|。|使⽤|My|SQL| 和|My|Batis|进⾏|数据|存|储|与|访问|，同时|可以|利⽤|Redis|缓存|来|进⼀步|提⾼|性 能|。|||上⾯讲过检索器不提供任何流式处理。但这⾥可以使⽤流式输出。我们可以把像 RAG 这样的链的执⾏过程想象成两个阶段：• 阶段⼀：准备阶段（同步、阻塞）：接收输⼊、检索⽂档、构建提⽰• 阶段⼆：⽣成阶段（可流式）：调⽤LLM流式输出的是最终答案，而非中间过程。
总结
我们可以总结一下
RAG 的完整流程可以分成离线准备和在线检索两大阶段，一共五个核心步骤
其实很简单就是
文档加载
文本分割
向量存储
检索
增强生成
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
26
0- 0
已为社区贡献2条内容
所有评论(0)