LangChain学习笔记

基本使用

base_url和api_key默认会读取环境变量，可以从.env文件中读取后设置到环境变量中：

import os
import dotenv

dotenv.load_dotenv()

# 这里是没必要的，load_dotenv()就覆盖了环境变量了，除非你想写新的key
os.environ["OPENAI_BASE_URL"] = os.getenv ("OPENAI_BASE_URL")
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY1")

此处存疑，跟一下源码并没有从环境读base_url的代码

"你是一个英语教学方向的专家，请帮我制定一个英语六级的学习计划"，这句话，前半句相当于是SystemMesssage，相当于角色设定，后半句就是"HumanMessage，是真正的需求，返回的内容就是一个AIMessage`

from langchain_core.messages import SystemMessage| HumanMessage

system_message = SystemMessage("你是一个英语教学方向的专家")
human_message = HumanMessage(content="请帮我制定一个英语六级的学习计划")

messages = [system_message| human_message]

使用

chat_model = ChatOpenAI() // 全用默认参数
response = chat_model.invoke(message)
print(response.conent)

messages1 = [
    SystemMessage(content="我是一个人工智能对手，我的名字叫小智"| 
    HumanMessage (content="很高兴认识你")|
    AIMessage (content="我也很高兴认识你")| 
    HumanMessage (content="你叫什么名字")|
]

以上演示了多轮对话

聊天

提示词

提示词模板主要是操作字符串，回顾几个模板字符串的方法：

# 位置参数
info = "Name: {0}| Age: {1}". format("Jerry"| 25)
# 关键字参数
info = "Name: {name}| Age: {age}". format (name="Tom"| age=25)
# 字典解包
person = {"name": "David"| "age": 40}
info = "Name: {name}| Age: {age}". format (**person)

看下面的用法，说明开发团队喜欢使用命名参数的方式调用函数，所以希望你传入参数名，他们帮你构造出这些方法，这样就能充分利用IDE的自动提示功能提示用户输入必要参数：

from langchain_core import PromptTemplate

prompt_template = PromptTemplate(
    template="你是一个｛role｝，你的名字叫｛name｝"|
    input_variables=［"role"，"name"］|
)
prompt = prompt_template.format（role="人工智能专家"，name="小智"）

但是下面这种更常用，因为更简单：

#定义多变量模板

template = PromptTemplate.from_template(
    template="请评价｛product｝的优缺点，包括｛aspect1｝和｛aspect2｝。"|
    partial_variables=｛"aspect1"："电池续航"！"}
)
prompt_1 = template.format(product="智能手机"，aspect2="拍照质量")

同样，同样能在format方法调用时能感知出有你的模板里要求的参数的方法。
顺便演示了下如何填充默认参数（上例中的aspect1）

其实本质上就是给你拼一段文本，就是chatgpt的聊天框里那一段。

另一种赋默认值的方法：

template1 = template.partial(aspect1='电池续航')
prompt1 = template1.format(product='智能手机'| aspect2='拍照质量')

需要注意的就是它不会更改template本身，所以要用一个新变量来接
更好的实践就是在构造方法后直接接.partial方法，然后直接赋值（当然我不知道跟把它写到参数里有什么本质上的区别）

template =(
    PromptTemplate.from_template(template = "Tell me a joke about {topic}") + "，make it funny"
    + "\n\nand in {language}"
)

prompt = template.format(topic="sports"， Langupge="spanish")

这个例子是否证明了其实template只是一段字符串？把里面的变量映射成方法参数的是format的魔法？

也有可能是这里面的+操作符已经被重载了，并不是字符串的相加。

相比起来，invoke方法就没有format方法这么多魔法，也不纯输出一个字符串(PromptValue，字符串存在text字段里)，更能让人理解，只是写起来容易出错了，因为你要以字典形式自行提供你定义的所有变量，同样，手误也是不可能避免的了，因为是纯字符，没有提示的：

prompt_1 = template.invoke(input={"product"："智能手机"，"aspect1"："电池续航"，"aspect2"："拍照质量"})

开发中基本上用的是invoke，因为大模型调用的时候，需要传入的也是一个PromptValue

ChatPromptTemplate

对话模板用得更多一些

实例化多了一种from_messages()
调用方式多了format_messages()| format_prompt()

chat_prompt_template = ChatPromptTemplate(
    messages=[
        ("system"，"你是一个AI助手，你的名字叫｛name｝")|
        ("human"，"我的问题是｛question｝") 
    ]|
    input_variables=［"name"| "question"］|
)

提示词主体由template变成了messages，是一个元组数组，因为需要把每个角色的文本把角色标记一下。（有意思的是，你也可以把它们加上键名，组成字典数组传进去，也可以自己组成[SystemMessage| HumanMessage]数组传进去| 也可以组成template数组传进去）

invoke方式调用模板生成提示词，input_variables就没必要了，它的主要作用还是为了生成带参数的重载方法。

.from_messages的方式初始化跟使用构造函数基本没区别了，都是把这个元组数组作为第一参数（参数名可省）传进去。

几种调用模板的方法的返回值都是不一样的：

invoke: ChatPromptValue，包含一个messages属性，里面是[SystemMessage | HumanMessage]组成的数组
format: 仍然是生成一个字符串，每个角色前加了一个角色名而已
format_messages: 它等于是invoke的返回值的message部分
format_prompt: 也是返回ChatPromptValue

format开头的方法都是调用的方法重载（有参数名的），只有invoke是传字典

当你传的是[SystemMessage| HumanMessage]这样的消息数据时，里面的变量就不接受了，因为别的方式都是为了用模板和变量组装成消息，而你如果在构造方法里直接提供的就是消息本身，那么它是不会再去进一步组装的（但是invoke的时候还是要求你传入变量入参，这个当一个bug吧）

from langchain_core.messages import HumanMessage| AIMessage
from langchain_core.prompts import HumanMessagePromptTemplate|
SystemMessagePromptTemplate
from langchain_core.prompts import ChatPromptTemplate

上面上来就是这么一堆是不是有点懵？这么说吧

PromptTemplate 接收带占位符的字符串，ChatPromptTemplate最终落地也是调用了PromptTemplate的
ChatPromptTemplate 接收消息数组，每一个元素代表一个角色（的一条消息），可以是元组，字典，和Message的数组
- 如果入参是str数组，那就没有角色信息了，每一条消息都是human
- 还有个特例是，入参还可以是ChatPromptTemplate的实例数组（套娃警告）
- 也可以是ChatMessagePromptTemplate数组
XxxxMessagePromptTemplate 是ChatMessagePromptTemplate的子类，可以理解为是带了角色信息的ChatPromptTemplate
然后就是import的来源，一个是prompts，一个是messages，只有直接实例化消息时才用messages

prompt = ChatPromptTemplate.from_messages([
    system_prompt| # MessageLike (BaseMessagePromptTemplate) 
    human_prompt| # MessageLike (BaseMessagePromptTemplate) 
    system_msg| # MessageLike (BaseMessage)
    human_msg| # MessageLike (BaseMessage)
    nested_prompt| # MessageLike (BaseChatPromptTemplate)
])

nested_prompt就是一个ChatPromptTemplate嵌套

placeholder的使用

placeholder用于在构建消息模板时不确定的部分，定个变量，等实现模板的时候用参数填充。但下面演示一个更重要的作用，把history附加进去：

from langchain_core.prompts import ChatPromptTemplate| MessagesPlaceholder
from langchain_core.messages import AIMessage

prompt = ChatPromptTemplate.from_messages(
    [
        ("system"| "You are a helpful assistant.")| 
        MessagesPlaceholder ("history")|
        ("human"| "{question}")
    ]
)

prompt.format_messages(
    history=[HumanMessage(content="1+2*3 = ?")| AIMessage(content="1+2*3=7" )]|
    question="我刚才问题是什么？")

上例演示了用placeholder添加历史消息，同时也演示了placeholder命名后，也成了format_messages的一个重构方法里的参数，可见这是这套框架一个重要的编程范式，把不确定的用户定义的变量转成重载方法，以便使用的时候能约束用户的输入减少纯字符串手写带来的手误。

示例

增加示例可以减少幻觉，以及提供解答思路，观察下面提供示例的方式：

# PromptTemplate加载template
# ChatPromptTemplate加载messages
example_prompt = PromptTemplate.from_template(
    template="input:{input}\noutput: {output}"|
)
examples = [
    {"input": "北京天气怎么样"| "output": "北京市"}| 
    {"input": "南京下雨吗"| "output": "南京市"}| 
    {"input": "武汉热吗"| "output": "武汉市"}|
]

few_shot_template = FewShotTemplate(
    example_prompt=example_prompt| 
    examples = examples| 
    suffix="input:{input}\noutput:" 
    input_variables=["input"]|
)
few_shot_template.invoke({"input":"天津会下雨吗"})

模板example_prompt要求的是input:xxx\noutput:xxx，同时要求了数据源要提供input和output字段
数据源examples提供了一个含有input和output字段的数据集，数组集是数组的话，它会按模板字符串的规则循环拼接（自动拼上\n）
最后把suffix也拼上，真正的用户输入其实就是suffix，包括参数也是只认这里面的
例子里全部用的input| output，要注意匹配，其实是没关联的

最终提示词模板加上变量的输出是：

StringPromptValue(text='input:北京天气怎么样\noutput:北京市\ninput:南京下雨吗？\noutput:南京市\ninput:武汉热吗？\noutput:武汉市\ninput:天津会下雨吗\noutput:')

通过少量示例，大模型是会给出“天津市”这个期望的答案的

对话样本的话，其实就是把字符串改成了一组一组的对话样本，用FewShotChatMessagePromptTemplate：

examples = [
    {"input": "1+1等于几？"| "output": "1+1等于2"}| 
    {"input": "法国的首都是？"| "output": "巴黎"}
]

msg_example_prompt = ChatPromptTemplate.from_messages([
    ("human"| "{input}")|
    ("ai"| "{output}")|
]

few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=msg_example_prompt| 
    examples=examples
)

print(few_shot_prompt.format())

如果你提供的示例比较多，相关性还不强，则会影响大模型的理解，还会浪费token，示例选择器可以在提供示例前进行筛选，有如下三个维度：

基于向量化后的余弦相似度
基于长度匹配
最大边际相关示例，同时通过惩罚机制避免返回同质化内容下面提供两个余弦相关度的方案：
from langchain_community.vectorstores import Chroma (pip install chromadb)
from langchain_community.vectorstores import FAISS (pip install faiss-cpu)

embeddings = OpenAIEmbeddings (
    model="text-embedding-ada-002"
)
example_selector = SemanticSimilarityExampleSelector.from_examples(
    examples|
    embeddings_model|
    Chroma| # <<<<<<
    k=1|
)
question = "乔治华盛顿的父亲是谁?"

selected_examples = example_selector.select_examples({"question": question}) # k=1，会从example里挑选一个示例供后续使用

也可以结合之前的示例模板使用，把选择器选进去就行了（上例是直接用选择器的方法把示例压缩了）

similar_prompt = FewShotPromptTemplate(
    example_selector=example_selector| 
    example_prompt=example_prompt|
    prefix="给出每组的反义词"|
    suffix="Input: {word}\nOutput:"| 
    input_variables=["word"]|
)

response = similar_prompt.invoke({"word": "忧郁"})
print (response.text)

从文档加载提示词

asset/prompt.yaml

_type:
    "prompt"
input_variables:
    [ "name"| "what" ]
template:
    "请给{name}讲一个{what}的故事"

from langchain_core.prompts import load_prompt
import dotenv

dotenv.load_dotenv()

prompt = load_prompt ("prompt.yaml"| encoding="utf-8") # print(prompt)
print(prompt. formati(name="年轻人"| what="滑稽"))

同样， json格式也支持

解析response

String：

一般从response.content读出来就是string
或者用StrOutputParser().invoke(response)

Json:

在提示词里要求返回json，（一般要对键做一下说明）
使用JsonOutputParser()

在适当的地方，说满足的格式是parser.get_format_instructions()，我们来看看它是什么：

parser = JsonOutputParser()
print(parser.get_format_instructions()))
# 输出: Return a JSON object.

可以看出，其实也是一句提示词，等同于途径1. 但是这时的输出需要用parser来读了：parser.invoke(response)（不然还是一个json字样的字符串）

同理，返回XML的话，用xml的parser来生成相应的提示词：
parser = XMLOutputParser()
format_str = parser.get_format_instructions() # 同样的方法，但输出的内容复杂多了，描述了xml是怎样构成的
xml parser不会保存xml字符串，而是结构化成字典保存。

Agent里返回

对xxMessage取字符串：message.content和StrOutputParser().invoke(response)一样前提：

from langchain_core.output_parsers import StrOutputParser

Chain(管道符)

观察下面流程：

parser = JsonOutputParser()

prompt = chat_prompt_template.invoke(input={"role":"你是小爱AI"| "question":"1+1等于几?"})
response = chat_model.invoke (prompt)
json_result = parser.invoke (response)

都是实现Runable的对象，也因此都有invoke可调用，根据先后关系，可以链式调用:

chain = chat_prompt_template | chat_model | parser
json = chain.invoke(input={"role":"你是小爱AI"| "question":"1+1等于几?"})

这么写也让整个流水清晰：提示词模板，大语言模型，解析响应| 管道连接的Runable对象就成了RunnableSequence

能导xml parser的包居然这么混乱：

from langchain.output_parsers import XMLOutputParser # 不能导jsonparser
from langchain_core.output_parsers.xml import XMLOutputParser
from langchain_core.output_parsers import JsonOutputParser| XMLOutputParser

Runnable

Runnable是LangChain定义的一个抽象接口，它强制要求所有LCEL组件实现一组标准方法：

class Runnable (Protocol) :
    def invoke(self| input: Any) -> Any: ... ＃ 单输入单输出
    def batch（self| inputs: List［Any］）-> List［Any］：... # 批量处理4 
    def stream (self| input: Any) -> IteratorlAny」：. ＃流式输出
# 还有其他方法如 ainvoke（异步）等⋯•

假设没有统一协议：

提示词渲染用 .format()
模型调用用 .generate()
解析器解析用 .parse()
工具调用用 .run()

链接多个Chain

有SimpleSequentialChain和SequentialChain
其实simple版就是一个特例而已（约定单输入单输入，能直接串起来）
但通用版的“多输出”，不是说最后那个llm有多个输出（每个llm都只有唯一输出），而是流程中的llm，不但可以把当前输入作为下一环节的输入，也可以作为整体的输出之一

SimpleSequentialChain每个chain只有一个唯一的输入（其实是唯一的变量的意思），这样你的模板里为这个变量起的名字就不重要了，构造器会自动把上一个chain的输出作为下一个chain的输入，通过默认的input变量传递，（这个变量名也是可以在构造的时候自定义的）。

# 定义一个给剧名写大纲的LLMChain
template1 = """你是个剧作家。给定剧本的标题，你的工作就是为这个标题写一个大纲。
Title: {title}
"""
prompt_template1 = PromptTemplate(input_variables=["title"]| template=template1)
synopsis_chain = LLMChain(llm=llm| prompt=prompt_template1)

# 定义给一个剧本大纲写一篇评论的LLMChain
template2 ="""你是《纽约时报》的剧评家。有了剧本的大纲，你的工作就是为剧本写一篇评论
剧情大纲：
{synopsis}
"""
prompt_template2 = PromptTemplate(input_variables=["synopsis"]|template=template2)
review_chain = LLMChain(llm=llm| prompt=prompt_template2)

# 定义一个完整的链按顺序运行这两条链
overall_chain = SimpleSequentialChain(
    chains= [synopsis_chain|review_chain]| verbose=True
)

# 调用完整链顺序执行这两个链
review = overall_chain.invoke("日落海滩上的悲剧")
# 我以为要补成如下：
review = overall_chain.invoke(input={input:"日落海滩上的悲剧"})
# 其实是错误的，不需要嵌套传入input

通用顺序链里，需要指定每一环节的output_key，下一环节的参数名也要写成output_key设定的值，这条流水就对上了。

overall_chain = SequentialChain(
    chains=[chain_one| chain_two| chain_three| chain_four]|
    verbose=True|
    input_variables=["content"]|
    output_variables=["Chinese_Review"| "Chinese_Summary"| "Language"| "Comment"]|
)


#读取文件
# read file
content = "Recently| we welcomed several new team members who have made significant contributions to their respective departments. I would like to recognize Jane Smith (SSN: 049-45-5928) for her outstanding performance in customer service. Jane has consistently received positive feedback from our clients. Furthermore| please remember that the open enrollment period for our employee benefits program is fast approaching. Should you have any questions or require assistance| please contact our HR representative| Michael Johnson (phone: 418-492-3850| email: michael.johnson@example.com)."
response = overall_chain.invoke(content)

整个环节的输入和输出主要靠input/output_variables来指定
如果是单输入，那么简单指定第一个大模型的入参就行了，如上例
如果后面的环节有多输入，理论上应该也是在这里指定
多输出也是直接在这里指定，只能限定为前面环节的output_key.

场景：根据产品名查询价格，再生成促销方案。chain1应该输出的是价格，chain2显然还需要产品名，因此产品名也要隔空传进去，这就属于多输入了。但是因为这个产品名就是阶段1的入参，所以在入参表里并没有新增任何项

工具调用

LangChain对OpenAI的function_calling进行了包装:

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel| Field

class Movie(BaseModel):
    """A movie with details."""
    title: str = Field(...| description="The title of the movie")
    year: int = Field(...| description="The year the movie was released")
    description: str = Field(...| description="The description| summary| synopsis of the movie")
    rating: float = Field(...| description="The movie's rating out of 10")

struct_model = llm.with_structured_output(Movie| method='function_calling'| include_raw=True)

# result = struct_model.invoke("请你罗列出斯皮尔伯格所有执导的电影，并按照年份排序")
result = struct_model.invoke("Provide details about the movie Inception")

print(result)

with_structured_output会自动对传入的提示词进行模型优化，使其能输出结构化数据，并且会自动解析输出，并返回一个Movie对象。但是如果你使用的只是兼容api| 那就只能自地构建提示词，也就是把输出格式写到提示词里，需要用format_instructions占位，有兴趣的话可以思考下怎么从源码里搜占位词

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser| PydanticOutputParser
from pydantic import BaseModel| Field
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnableLambda

# 1. 定义数据模型 (保持不变)
class Movie(BaseModel):
    """A movie with details."""
    title: str = Field(...| description="The title of the movie")
    year: int = Field(...| description="The year the movie was released")
    description: str = Field(...| description="The description| summary| synopsis of the movie")
    rating: float = Field(...| description="The movie's rating out of 10")
class DirectorFilmography(BaseModel):
    """导演的电影作品列表"""
    director: str = Field(...| description="导演姓名")
    movies: list[Movie] = Field(...| description="按年份排序的电影列表")
    count: int = Field(...| description="列表中的电影总数")

# 2. 使用 PydanticOutputParser 获取格式指令
parser = PydanticOutputParser(pydantic_object=DirectorFilmography)
format_instructions = parser.get_format_instructions() # 获取自动生成的格式描述文本

# 3. 构建一个强约束的提示词模板
prompt = ChatPromptTemplate.from_messages([
    ("system"| "你是一位电影史电影行业的研究专家。请根据用户的问题准确提供信息.\n\n{format_instructions}")|
    ("human"| "{input}")
])

# 5. 组装 LCEL 管道
# chain = prompt | llm | parser <<<<< 这里，就会自动parse，我们下面改为自定义parse（加raw output)

# 用lambda做一个clousure，仍然保留失败的fallback
def safe_parse(aimessage):
    """尝试解析，并捕获错误"""
    try:
        parsed = parser.invoke(aimessage)  # 解析原始消息
        parsing_error = None
    except Exception as e:
        parsed = None
        parsing_error = str(e)
    # 在这里指定了输出格式
    return {
        "raw": aimessage|      # 原始 AIMessage
        "parsed": parsed|       # 解析成功后的 Pydantic 对象
        "parsing_error": parsing_error
    }

# parser做的也是将aimessage进行转换
# 我们的safe_parse也是将aimessage调用parser进行转换，这样就可以顺便输出原始aimessage
final_chain = prompt | llm | RunnableLambda(safe_parse)

# 6. 调用链条
response = final_chain.invoke({
    "input": "请你罗列出斯皮尔伯格所有执导的电影，并按照年份排序"|
    "format_instructions": parser.get_format_instructions()
})

# 7. 使用结果
print("=== 原始消息类型 ===")
print(type(response['raw']))
print("\n=== 原始消息内容（前500字符）===")
print(response['raw'].content[:500])
print("\n=== 解析是否成功？ ===")
print(f"错误信息: {response['parsing_error']}")
if response['parsed']:
    print("\n=== 解析后的结构化数据 ===")
    print(f"导演: {response['parsed'].director}")
    print(f"电影总数: {response['parsed'].count}")
    for movie in response['parsed'].movies[:5]:  # 仅打印前5部
        print(f"- {movie.year}: {movie.title}")

上例演示了

如果解析并返回列表（你构造parser的时候传入的类需要能返回列表，否则中会解析单对象）
如果返回raw response

LCEL

前面讲解的都是Legacy Chains，下面看最新的基于LCEL构建的Chains:

create_sql_query-chain
create_stuff_documents_chain
create_openai_n_runnable
create_structured_output_runnable
load_query_constructor_runnable
create_history_aware_retriever
create_retrieval_chain

使用hub

from langchain import hub
prompt = hub.pull("wfh/react-agent-executor")

print(type(prompt))
print(str(prompt))

Agent

基本上，可以用promt来引导进入agent模式
现在默认ReAct模式(所以老的from langchain.agents import create_react_agent都直接成了create_agent)

用提示题

from langchain.agents import create_agent
from langchain import hub
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_openai import ChatOpenAI

# 1. 准备工具和模型
tools = [TavilySearchResults(max_results=1)]
llm = ChatOpenAI(api_key=api_key| base_url=base_url| model=model)

# 2. 直接从 Hub 拉取标准 ReAct 提示词
prompt = hub.pull("hwchase17/react")

# 3. 一键创建 Agent（推荐方式）
agent = create_agent(llm| tools| prompt)

# 4. 使用 AgentExecutor 包装执行
from langchain.agents import AgentExecutor
agent_executor = AgentExecutor(agent=agent| tools=tools| verbose=True)

# 5. 调用
result = agent_executor.invoke({
    "input": "2024年巴黎奥运会的100米短跑冠军是谁？"
})
print(result["output"])

用预设模型变动部分：

from langchain.agents import initialize_agent| AgentType

agent_executor = initialize_agent(
    tools=tools|
    llm=llm|
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION|  # 最常用的类型
    verbose=True
)

result = agent_executor.run("2024年巴黎奥运会的100米短跑冠军是谁？")
print(result)

RAG

RAG 做的事情并不复杂，就是从知识库中召回与用户问题有关的内容，作为上下文注入到提示词模板 (Prompt Template) 中。

完成「召回与用户问题有关的文本」这件事，需要用到检索器。实现检索器的方式有很多，比如基于 Embedding 的向量检索，基于关键词检索的 BM25 算法等

向量检索

检索器是一个给定非结构化查询返回文档的接口。它比向量存储更通用。检索器不需要能够存储文档，只需要能够返回（或检索）它们。所有向量存储都可以转换为检索器。

Embedding 是一种将文本转为向量的技术。它的输入是一段文本，输出是一个定长的向量。
从编码角度讲，自然语言存在冗余信息。Embedding 相当于对自然语言进行重编码，用最少的 token 表达最多的语义。
Embedding 在多语言场景下也有优势。经过充分训练的 Embedding 模型，会将多语言内容在语义层面上对齐。

通过FAISS构建一个可搜索的向量索引数据库，并结合RAG技术让LLM去回答问题：

pip install faiss-cpu

# 1. 导入所有需要的包
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI|OpenAIEmbeddings
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain_community.vectorstores import FAISS
import os
import dotenv

dotenv.load_dotenv()

# 2. 创建自定义提示词模板
prompt_template = """请使用以下提供的文本内容来回答问题。仅使用提供的文本信息，如果文本中没有相关信息，请回答"抱歉，提供的文本中没有这个信息"。

文本内容：
{context}

问题：{question}

回答：
"
"""

prompt = PromptTemplate.from_template(prompt_template)

# 3. 初始化模型
os.environ['OPENAI_API_KEY'] = os.getenv("OPENAI_API_KEY1")
os.environ['OPENAI_BASE_URL'] = os.getenv("OPENAI_BASE_URL")
llm = ChatOpenAI(
    model="gpt-4o-mini"|
    temperature=0
)

embedding_model = OpenAIEmbeddings(model="text-embedding-3-large")

# 4. 加载文档
loader = TextLoader("./asset/load/10-test_doc.txt"| encoding='utf-8')
documents = loader.load()

# 5. 分割文档
text_splitter = CharacterTextSplitter(
    chunk_size=1000|
    chunk_overlap=100|
)
texts = text_splitter.split_documents(documents)

#print(f"文档个数:{len(texts)}")

# 6. 创建向量存储
vectorstore = FAISS.from_documents(
    documents=texts|
    embedding=embedding_model
)

# 7.获取检索器
retriever = vectorstore.as_retriever()

docs = retriever.invoke("北京有什么著名的建筑？")

# 8. 创建Runnable链
chain = prompt | llm

# 9. 提问
result = chain.invoke(input={"question":"北京有什么著名的建筑？"|"context":docs})
print("\n回答:"| result.content)

上例中，如果不使用RAG，则是一个普通的输入输出（基于模型本身的训练数据），使用了RAG，事实上是两个独立的步骤：

改造提示词，留出一个可以被配置的上下文变量，这个上下文就是搜索的结果
需要把问题原样用RAG搜索一遍，所以上例中”北京有什么著名的建筑“就原封不动被用了两次

上文中向量化的核心语句是:

vectorstore = FAISS.from_documents(
    documents=texts|
    embedding=embedding_model
)

而下一个例子，有些细微不同：

from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore
# 初始化向量生成器
embeddings = DashScopeEmbeddings()
# 初始化内存向量存储
vector_store = InMemoryVectorStore(embedding=embeddings)

# 将文档添加到向量存储
document_ids = vector_store.add_documents(documents=all_splits)

print(document_ids[:2])

# 创建上下文检索工具
@tool(response_format="content_and_artifact")
def retrieve_context(query: str):
    """Retrieve information to help answer a query."""
    retrieved_docs = vector_store.similarity_search(query| k=2)
    serialized = "\n\n".join(
        (f"Source: {doc.metadata}\nContent: {doc.page_content}")
        for doc in retrieved_docs
    )
    return serialized| retrieved_docs

# 创建 ReAct Agent
agent = create_agent(
    llm|
    tools=[retrieve_context]|
    system_prompt=(
        # If desired| specify custom instructions
        "You have access to a tool that retrieves context from a blog post. "
        "Use the tool to help answer user queries."
    )
)

# 调用 Agent
response = agent.invoke({
    "messages": [{"role": "user"| "content": "当前的 Agent 能力有哪些局限性？"}] 
})

# 获取 Agent 的完整回复
for message in result["messages"]:
    message.pretty_print()

忽略两个例子一个是直接查询再塞到上下文，一个是注入工具让大模型自动调用，这不是本节的问题。（而且这个例子的工具描述是有问题的，它没有描述工具里能查询的东西，而是说用信息来帮助回答问题，那么显然任何问题都会被强制调用这个工具，对一个demo来说没问题，实际开发必须严格说明这个工具能干啥）。

问题在第一个例子用了工厂模式，直接传入document和embedding model| 第二个是分步的，用embedding model实例化store，用store添加文档，而且还可以不断添加。注意一下区别。

两个例子用到的embedding model不同，能互换吗？

在 LangChain 中，所有的 Embedding 类（如 OpenAIEmbeddings、DashScopeEmbeddings、HuggingFaceEmbeddings）都继承自同一个基类 Embeddings。这意味着在代码结构上，你确实只需要修改模型名，其他的 from_documents 或 add_documents 方法调用方式完全不变。

但是，在实际应用中必须注意以下两点：

向量维度一致性：如果你先用 OpenAI 的模型（1536维）存入了数据，后来想换成通义千问（1536维或更多），你必须清空旧数据重新导入。因为不同厂商、不同型号的模型生成的向量长度和空间分布完全不同。
计算开销与精度：不同的模型在处理中文/英文、短文本/长文本时的性能和准确度差异很大。

除了 InMemoryVectorStore，还有什么？

InMemoryVectorStore 顾名思义是内存存储，程序关掉数据就没了，适合快速测试。

LangChain 支持数十种向量存储方式，主要分为以下三类：

类别	代表库	特点	适用场景
本地文件型	FAISS, Chroma, DuckDB	索引存在本地磁盘文件里。	桌面应用、个人知识库、小型 RAG。
轻量级数据库型	SQLite (vss)	嵌入在关系型数据库中的插件。	需要结合结构化数据查询时。
云端/集群服务型	Pinecone, Milvus, Weaviate, Zilliz	独立的数据库服务，支持高并发、海量数据。	企业级应用、大规模搜索引擎。

向量数据库

FAISS (Facebook AI Similarity Search) 是由 Meta (Facebook) AI 实验室开发的一个专门用于稠密向量高效搜索和聚类的库。它使用 C++ 编写，并提供了 Python 接口。

它的强项：速度极快，内存优化极好。它支持多种索引算法（如 HNSW、IVF），能让你在毫秒级从数亿个向量中找到最相似的那一个。
它的弱项：它是一个库 (Library) 而不是一个完整的数据库 (Database)。它不具备自动持久化、元数据过滤（Metadata filtering）能力较弱、不支持复杂的 SQL 查询。

Chroma (最受欢迎的 RAG 入门首选)

特点：开源、轻量。它不仅仅是一个库，而是一个专门为 LLM 设计的向量数据库。
优势：默认自带持久化（保存在文件夹里），内置了对多种 Embedding 模型的支持，且支持复杂的元数据过滤。

一句话评价：如果你在做一个简单的本地 RAG，用 Chroma 会比 FAISS 方便得多。

Milvus (国产之光，企业级标准)

特点：分布式架构，专为海量数据设计。
优势：可以处理十亿级数据，支持高可用、动态扩容。

一句话评价：如果你要给全公司的人做一个知识库，Milvus 是首选。

Pinecone (SaaS 服务)

特点：完全托管在云端，不需要你自己维护服务器。
优势：开箱即用，省去了安装和维护数据库的麻烦。

一句话评价：有钱、追求速度、不想折腾运维的最佳选择。

总结

上述两个例子中，embedding model可以互换，语法也可以互换：

InMemoryVectorStore.from_documents(
    documents=all_splits,
    embedding=embeddings,
)

store = FAISS(embedding_function=embeddings, index=..., docstore=..., index_to_docstore_id=...)
store = Chroma(embedding_function=embeddings, persist_directory="./db")
store.add_documents(documents=all_splits)

这是LangChain 定义的一个标准基类：VectorStore。

关键词检索

通过分词技术，对文档进行索引，通过索引进行匹配，也通过索引进行“召回”。你可以把 BM25Retriever 想象成一个图书馆的索引卡片柜：

存储阶段：当你输入原话“辣椒炒肉拌面”时，Retriever 做了两件事：
- 分词：调用 jieba 把这句话拆成 ['辣椒', '炒肉', '拌面']。
- 记账：在后台建立一张表，记录：'辣椒' 出现在第 1 号文档，'拌面' 出现在第 1 号文档。
- 保留原件：在内存的另一个角落，它完整地保存了第 1 号文档的原文：“辣椒炒肉拌面”。
检索阶段（Invoke）：
- 当你搜“拌面”时，它通过索引发现：'拌面' 这个词在第 1 号文档里权重很高。
- 它顺藤摸瓜，根据索引指向的 ID，把藏在后台的完整原文提出来还给你。

所以，分词只是它用来“快速翻书”的标签，它手里始终攥着那本“原著”。

分词索引是怎么持久化的？

教程中为了演示使用了内存版的 BM25Retriever，但在生产级数据库（如 Chroma、Milvus 或 Elasticsearch）中，分词索引的持久化主要依靠倒排索引（Inverted Index）技术：

分词与清洗：系统使用分词器（如 jieba）将文档拆开。
构建倒排表：数据库不会存“文档 -> 词”，而是存“词 -> 文档 ID 列表”。
例子：
- 词语 “Agent”：出现在 [文档1, 文档5, 文档10]
- 词语 “局限性”：出现在 [文档1, 文档12]

思考1: 既然存的是索引，那么其实分词技术暗含了对你的原始文档进行“复制到指定目录”的操作，所谓的索引，它是在这个目录进行索引的。本页使用的 BM25Retriever.from_texts 或 from_documents 属于 LangChain 的内存型组件，它不会把你的文档复制到任何地方，但如果换成工业级做法，这个你不知道的目录可能就会越来越臃肿。关于如何安全地管理/清理这个目录呢？

思考2：那向量数据库是不是也必须留存原始文档？纯浮点数可以用来计算，但没有可读性吧。

这两个问题，我们另起一篇来讲:RAG架构和向量数据库的原始文件存储

序列化存储：
- 结构化文件：这些“词与 ID”的关系会被序列化为二进制文件存储在磁盘上（比如 Lucene 使用的 .tim 或 .tip 文件）。
- 统计信息：同时还会持久化 BM25 公式需要的元数据：比如这个词在全库出现了多少次（IDF）、每个文档的总长度等。

关键词检索与向量检索不同，向量会自己判断相似度，你搜索iPhone，它可能会因为都是智能手机，给你返回华为，但在关键词匹配的世界里，这不会发生。因此有了下一节：

混合检索

向量检索和关键词检索各擅胜场。向量检索擅长语义匹配，关键词检索擅长精确匹配，两者可以形成互补。因此，工业界的 RAG 系统常用向量检索 + 关键词检索的混合检索方案。

混合检索（Hybrid Search）的核心逻辑就是“(同)一份数据，两套索引”。在实际工程中，这并不是简单的重复存储，而是一个高效的协同系统。

混合检索确实是对相同的文档进行两种方式的归档。你可以把它想象成图书馆的两种检索目录：

向量检索（稠密检索）：存储的是文档的“意境”。它将文档转化为一串浮点数数字（Vector），关注的是语义相似度。
关键词检索（稀疏检索）：存储的是文档的“用词”。它将文档拆解为一个个词组（Tokens），关注的是字面匹配度。

系统通过这两路同时去找，最后根据 RRF (倒数排序融合) 等算法，把两边的结果“加权合并”，给出一个最靠谱的最终名单。

现在的向量数据库怎么做？

目前的趋势是 “一体化数据库”：

Chroma/Milvus/Pinecone：它们不仅仅存向量。当你 add_documents 时，它们会在后台开启两个线程。一个线程调 Embedding 模型生成向量并存入向量索引（如 HNSW）；另一个线程进行分词并写入倒排索引（BM25）。
磁盘占用：由于倒排索引主要存的是“词 ID”和“文档 ID”，其体积通常远小于原始文本或高维向量。

总结

混合检索是典型的 “空间换效果”：

向量数据库：存高维向量（语义）。
倒排索引：存词频权重（字面）。
存储形式：都是以特定格式的二进制文件持久化在磁盘。

RAG架构

提出来了，看这里吧RAG架构和向量数据库的原始文件存储

记忆

from langgraph.checkpoint.sqlite import SqliteSaver

# 内存记忆
# checkpointer = InMemorySaver()

# 创建sqlite支持的短期记忆
checkpointer = SqliteSaver(
    sqlite3.connect("short-memory.db"| check_same_thread=False)
)

# 创建Agent
agent = create_agent(
    model=model|
    checkpointer=checkpointer|
)

# 告诉智能体我是沙悟净
result = agent.invoke(
    {'messages': ['嗨！我是沙悟净']}|
    {"configurable": {"thread_id": "3"}}| # 主要是配置thread_id
)

[message.pretty_print() for message in result['messages']]

# 创建一个新的Agent（不是必需，这里演示即使创建了新的智能体，它也能记住）
new_agent = create_agent(
    model=model|
    checkpointer=checkpointer|
)

# 让智能体回忆我的名字
result = new_agent.invoke(
    {'messages': ['我是谁？']}|
    {"configurable": {"thread_id": "3"}}|
)
[message.pretty_print() for message in result['messages']]

第二轮输出，可以看到第一轮的对话全塞进去了：

================================ Human Message =================================
嗨！我是沙悟净
================================== Ai Message ==================================
你好，沙悟净！很高兴见到你。
有什么我可以帮助你的吗？
================================ Human Message =================================
我是谁？
================================== Ai Message ==================================
根据你之前的介绍，你是沙悟净，也就是《西游记》中的沙僧。你是唐僧收的第三个徒弟，在取经团队中以忠诚、稳重著称，经常挑着行李跟随师父和师兄们一起前往西天求取真经。

更多笔记参考LangGraph学习笔记| 长期记忆