langchain是什么?是一个类库吗?
不是，langchain是一套开源框架， 相当于java中的springboot框架。
也就是说，许多重复性的工作不用做了，例如连接数据库，只需要一行代码搞定。

langchain很简单吗?当然不是，就问一个最简单的问题，langchain1.2有哪些新机制?
这可是区分高手和菜鸟的分水岭。见下文。

入门示例

环境准备

python #3.11

步骤

1、卸载依赖(避免冲突)

pip install langchain>=0.3.0 langchain-core>=0.3.0 langchain-community>=0.3.0 langchain-huggingface>=0.1.0 chromadb sentence-transformers torch transformers                        

2、安装依赖

 pip install langchain>=0.3.0 langchain-core>=0.3.0 langchain-community>=0.3.0 langchain-huggingface>=0.1.0 chromadb sentence-transformers torch transformers                        

3、新建python文件，rag_demo，代码：

import os
import sys
from typing import Dict, Any

# --- 1. 导入最新 LangChain 组件 (v0.3+) ---
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda

# 向量库与嵌入
from langchain_community.vectorstores import Chroma
from langchain_huggingface import HuggingFaceEmbeddings

# 文本分割
from langchain_text_splitters import CharacterTextSplitter

# --- 2. 配置模型 ---
EMBED_MODEL_NAME = "BAAI/bge-small-zh-v1.5"


def mock_llm_func(input_data: Dict[str, Any]) -> str:
    """
    模拟 LLM 函数。
    注意：在 Chain 中，经过 prompt 后，输入通常是 ChatPromptValue。
    为了兼容，我们这里直接处理原始字典数据（在 prompt 之前拦截）或者调整链结构。

    【修正策略】：我们将模拟逻辑放在 prompt 格式化之后，解析 messages。
    但更简单的做法是：不让 prompt 输出 ChatPromptValue，而是直接输出字符串给模拟函数，
    或者让模拟函数能够处理 messages。

    这里采用最通用的方式：接收 messages 并提取用户问题上下文。
    """
    # input_data 在这里实际上是 ChatPromptValue 转换后的 messages 列表 (如果使用了 to_messages)
    # 或者如果是 RunnableLambda 直接接在 prompt 后，它收到的是 ChatPromptValue

    # 为了演示简单，我们改变链的结构：
    # 不在 prompt 后直接接 lambda，而是构造一个完整的字符串提示词给 lambda
    return f"[模拟AI回答] 基于信息：'{input_data.get('context', '无')}'，回答 '{input_data.get('question', '')}': LangChain 已成功运行！"


def main():
    print("🚀 正在初始化 LangChain 0.3+ (修复版)...")

    # --- A. 准备数据 ---
    raw_texts = [
        "LangChain 是一个用于开发由语言模型驱动的应用程序的框架。",
        "Python 3.11 带来了显著的性能提升，特别是对于异步操作。",
        "LangChain 0.3 版本进行了重大的架构重构，将社区集成移入了 langchain-community 包。",
        "HuggingFace 提供了数千种免费的预训练模型，包括嵌入模型和生成模型。",
        "RAG (检索增强生成) 技术通过结合外部知识库来提高大模型的准确性。"
    ]
    documents = [Document(page_content=text) for text in raw_texts]

    # --- B. 文本分割 ---
    text_splitter = CharacterTextSplitter(chunk_size=100, chunk_overlap=0)
    splits = text_splitter.split_documents(documents)

    # --- C. 初始化嵌入模型 ---
    print(f"⏳ 加载嵌入模型: {EMBED_MODEL_NAME} ...")
    try:
        embeddings = HuggingFaceEmbeddings(
            model_name=EMBED_MODEL_NAME,
            model_kwargs={"device": "cpu"},
            encode_kwargs={"normalize_embeddings": True}
        )
    except Exception as e:
        print(f"❌ 模型加载失败: {e}")
        return

    # --- D. 构建向量存储 ---
    vectorstore = Chroma.from_documents(documents=splits, embedding=embeddings)
    retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

    # --- E. 定义提示词模板 ---
    template = """基于以下已知信息回答问题。
    已知信息: {context}
    问题: {question}
    回答:"""
    prompt = ChatPromptTemplate.from_template(template)

    # --- F. 【关键修复】构建链 ---
    # 错误原因之前是：Prompt 输出 ChatPromptValue -> Lambda (期望 dict)
    # 修复方法：使用 RunnableLambda 处理 dict 输入，并在 Prompt 之前完成上下文组装，
    # 或者让 Lambda 处理 messages。

    # 方案：我们手动组装字符串传给模拟函数，绕过 ChatPromptValue 的复杂性
    # 这样最适合“模拟”场景，不需要真正的 Message 对象

    def format_docs(docs):
        return "\n\n".join([d.page_content for d in docs])

    def simple_mock_chain(input_dict):
        """
        完全手动的模拟链，不依赖 ChatPromptTemplate 的复杂对象，
        确保在没有任何 API Key 的情况下绝对可运行。
        """
        context = input_dict["context"]
        question = input_dict["question"]
        full_prompt = f"基于以下已知信息回答问题。\n已知信息: {context}\n问题: {question}\n回答:"
        # 模拟 LLM 思考
        return f"🤖 AI 回复: 根据资料 '{context[:20]}...', 关于 '{question}' 的答案是：LangChain 流程跑通了！"

    # 构建正确的链
    # 1. 检索文档并格式化为字符串
    # 2. 将 context 和 question 组装成字典
    # 3. 传入我们的模拟函数
    retrieval_chain = (
            {
                "context": retriever | format_docs,
                "question": RunnablePassthrough()
            }
            | RunnableLambda(simple_mock_chain)
    )

    # --- G. 执行测试 ---
    print("\n" + "=" * 50)
    test_questions = [
        "LangChain 是什么？",
        "Python 3.11 有什么特点？",
        "RAG 技术是如何工作的？"
    ]

    for q in test_questions:
        print(f"\n👤 用户: {q}")
        try:
            response = retrieval_chain.invoke(q)
            print(f"   {response}")
        except Exception as e:
            print(f"❌ 出错: {e}")
            import traceback
            traceback.print_exc()

    print("\n" + "=" * 50)
    print("✅ 演示成功！无 'ChatPromptValue' 错误。")


if __name__ == "__main__":
    os.environ["HF_ENDPOINT"] = "https://hf-mirror.com"
    try:
        main()
    except KeyboardInterrupt:
        print("\n🛑 中断。")

4、运行结果

🚀 正在初始化 LangChain 0.3+ (修复版)...
⏳ 加载嵌入模型: BAAI/bge-small-zh-v1.5 ...
==================================================

👤 用户: LangChain 是什么？
   🤖 AI 回复: 根据资料 'LangChain 是一个用于开发由语言...', 关于 'LangChain 是什么？' 的答案是：LangChain 流程跑通了！

👤 用户: Python 3.11 有什么特点？
   🤖 AI 回复: 根据资料 'Python 3.11 带来了显著的性能...', 关于 'Python 3.11 有什么特点？' 的答案是：LangChain 流程跑通了！

👤 用户: RAG 技术是如何工作的？
   🤖 AI 回复: 根据资料 'RAG (检索增强生成) 技术通过结合外...', 关于 'RAG 技术是如何工作的？' 的答案是：LangChain 流程跑通了！

==================================================
✅ 演示成功！无 'ChatPromptValue' 错误。

langchain

langchain的标准流程

加载 → 切分 → 向量化 → 存储 → 检索 → 回答
这就是 LangChain 最核心、最标准、最常用的构建流程。

1、 加载文件（Load）
把你的 PDF / Word / TXT 读进程序。
2、 切分文本（Split）
文件太长，AI 记不住，切成一段一段小文本。
3、 生成向量（Embedding）
把每段小文本变成数字向量（方便搜索）。
4、 存入向量库（Store）
把所有向量存在 Chroma 里（你装的那个库）。
5、 用户提问 → 检索相似内容（Retrieve）
用户问问题，系统去库里找最相关的几段文字。
6、 送给大模型 → 生成回答（Generate）
把问题 + 找到的资料一起给 AI

加载文件

加载文件-从字符串中加载

代码：

raw_texts = [
        "LangChain 是一个用于开发由语言模型驱动的应用程序的框架。",
        "Python 3.11 带来了显著的性能提升，特别是对于异步操作。",
        "LangChain 0.3 版本进行了重大的架构重构，将社区集成移入了 langchain-community 包。",
        "HuggingFace 提供了数千种免费的预训练模型，包括嵌入模型和生成模型。",
        "RAG (检索增强生成) 技术通过结合外部知识库来提高大模型的准确性。"
    ]
    documents = [Document(page_content=text) for text in raw_texts]

加载文件-从txt/md中加载

代码：

from langchain_community.document_loaders import TextLoader  # 导入文件加载器

# 直接加载文件（自动读取全部内容）
loader = TextLoader("你的文件.txt", encoding="utf-8")
documents = loader.load()  # 直接得到 Document 列表！

加载文件-从pdf中加载

先安装依赖：

pip install pypdf

代码：

from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader("你的文件.pdf")
documents = loader.load()

加载文件-从word中加载

先安装依赖：

pip install python-docx

代码：

from langchain_community.document_loaders import Docx2txtLoader

loader = Docx2txtLoader("你的文件.docx")
documents = loader.load()

切分文本

代码：

    text_splitter = CharacterTextSplitter(
    chunk_size=100, 
    chunk_overlap=0)
    splits = text_splitter.split_documents(documents)

切分文本参数

参数名	意思	常用值
chunk_size	每一段最大长度	200 / 500 / 1000
chunk_overlap	段落重叠多少字符	20 / 50 / 100
separator	按什么符号切割	\n（换行）、。（句号）

生成向量

代码：

vectorstore = Chroma.from_documents(
documents=splits, 
embedding=embeddings)

这一行代码做了两件事情：
1、根据切好的文本，生成向量
2、把向量存入Chroma数据库

所以说langchain的封装度很高。

生成向量-伪代码

代码：

# 伪代码（真实步骤，被隐藏了）
for 一段文本 in 切好的文本:
    向量 = embedding_model.generate_embedding(一段文本)  # 生成向量
    向量数据库.add(向量)

用langchain的话，一行代码全包了。

检索

代码：

retrieval_chain = (
            {
                "context": retriever | format_docs,
                "question": RunnablePassthrough()
            }
            | RunnableLambda(simple_mock_chain)
    )

调用代码，q是传入的问题：
retrieval_chain.invoke(q)

检索并不单独存在，是invoke方法中的一步。

以上代码两大步：
1、定义调用链
retriever # 从向量数据库检索
| # 管道符号，表示然后
format_docs # 将documents整理为一段文字
RunnableLambda(simple_mock_chain) # 自定义的串行函数
2、执行
invoke()才是发起执行。

format_docs为什么要将documents整理为文字呢?

举个例子就容易理解了，查出来的内容：

[
    Document(page_content="第一段内容"),
    Document(page_content="第二段内容"),
    Document(page_content="第三段内容")
]

整理后的内容：

第一段内容
第二段内容
第三段内容

回答

回答也并不单独存在，也包括在invoke方法中。

初始化嵌入模型

代码：

# --- C. 初始化嵌入模型 ---
    print(f"⏳ 加载嵌入模型: {EMBED_MODEL_NAME} ...")
    try:
        embeddings = HuggingFaceEmbeddings(
            model_name=EMBED_MODEL_NAME,
            model_kwargs={"device": "cpu"},
            encode_kwargs={"normalize_embeddings": True}
        )
    except Exception as e:
        print(f"❌ 模型加载失败: {e}")
        return

langchain的角色定位实体类

这些类是langchain_core包自带的。

类名	角色定位	谁发送的？	典型内容
SystemMessage	导演/剧本	开发者	“你是一个乐于助人的助手。”
HumanMessage	观众/客户	用户	“请帮我写一首诗。”
AIMessage	演员	AI 模型	“好的，这是为你写的诗…”
ToolMessage	道具组	外部工具/API	“当前时间是 12:00。”
AIMessageChunk	提词器	AI 模型 (流式)	“好…的…这…是…” (逐字)

langchain1.2有哪些新机制

1、LCEL声明式架构
强调统一协议(Runnable)带来的开发效率，支持一套代码无缝切换同步、异步、并发、流式处理。

方法调用主要有三种：
单次调用、批量调用、流式调用，这三种每种都有异步方法。

如下表：

调用类型	同步方法	异步方法	说明
单次调用	invoke(input)	ainvoke(input)	最常用，处理单个请求。
批量调用	batch(inputs)	abatch(inputs)	处理列表数据，效率更高。
流式调用	stream(input)	astream(input)	逐字输出（打字机效果），用于聊天界面。

2、弹性容错机制
Retry配合Fallback模型降级，构建具备“自愈”能力的生产系统，保证业务死不掉。
with_retry()
with_fallbacks()
3、流式事件透传
利用astream_events解决长链路的延迟焦虑，实时反馈中间过程，极大提升用户体验。
4、LangGraph 状态机
展现处理复杂循环逻辑的能力，利用 Checkpoint存档机制解决Agent运行状态丢失的痛点。

链式操作

链式操作是1.2版本之后才有的吗?

首先要明确这里的链式操作指什么?
主要有两个版本：
老链式(已过时)
LLMChain(llm=..., prompt=...) # 定义
.run()或.apply()等来调用

新链式(推荐)
chain = prompt | model # 定义
.invoke()、.stream()等来调用
新链式严格来说也不是1.2才有的，应该是1.1.7，不过归到1.2也没什么问题。

链式操作必须实现runnable接口吗?

是的。
这些常见的原生对象本身就已经实现了runnable接口：
PromptTemplate / ChatPromptTemplate
ChatModel / LLM
OutputParser
Retriever (检索器)
Tool (注意：这里指封装好的 Tool 对象，但直接用 | 连接时通常也需要它是 Runnable 格式，见下文)

其他

文档

langchain官网文档(quickstart页面)：
https://docs.langchain.com/oss/python/langchain/quickstart

国内中文文档：
https://langchain.ichuangpai.com/

国内中文文档-2：
https://www.langchain.com.cn/docs/introduction/