一、LangChain 基本使用

1.1 基础消息模型调用

核心消息类型：SystemMessage（系统角色）、HumanMessage（用户请求）、AIMessage（模型响应）

from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 初始化模型（Python关键字参数，config配置推荐环境变量，避免硬编码）
model = ChatOpenAI(
    model=”qwen-turbo”,  # 模型名称（Python字符串参数）
    # temperature：随机性0~2（Python浮点型），越高越发散
)

# 消息列表（Python列表，存储消息对象）
messages = [
    SystemMessage(“将英文翻译成通顺中文”),
    HumanMessage(“hi!”)
]

# 调用+解析（Python对象方法调用）
result = model.invoke(messages)  # invoke方法传入消息列表，获取响应
parser = StrOutputParser()       # 实例化解析器（Python类实例化）
print(parser.invoke(result))     # 解析响应并打印（Python打印函数）

1.2 LCEL 链式调用

借鉴Linux管道思想，| 运算符重载，串联可运行组件（Python运算符重载语法）

# 链式简写（| 管道符，串联模型与解析器）
chain = model | parser
ret = chain.invoke(messages)  # 链式调用，简化代码

# 等价写法（Python对象构造，两种方式）
from langchain_core.runnables import RunnableSequence
chain1 = RunnableSequence(first=model, last=parser)  # 关键字参数构造
chain2 = model.pipe(parser)  # 实例方法调用

1.3 统一模型初始化 init_chat_model

统一入口创建多厂商模型，支持运行时动态配置参数（Python字典传参语法）

from langchain.chat_models import init_chat_model

# 初始化可配置模型（Python关键字参数，指定可配置字段）
config_model = init_chat_model(
    model=”deepseek-chat”,
    temperature=1.0,
    configurable_fields=[“model”, “temperature”],  # 可动态修改的参数（Python列表）
    config_prefix=”llm”
)
# 调用时临时修改参数（Python嵌套字典传参）
res = config_model.invoke(
    “你好”,
    config={“configurable”: {“llm_temperature”: 0.3}}
)

二、LangChain 工具系统

2.1 工具作用

打破大模型知识截止、封闭无法联网的局限，扩展外部能力。

2.2 自定义工具

核心三要素：函数+类型注解+标准文档注释，自动生成工具Schema。

方案1：基础@tool装饰器 + Google风格文档注释（简单工具首选）

Python函数注解（指定参数/返回值类型）、文档字符串（说明函数用途与参数）

from langchain_core.tools import tool

# @tool装饰器：将普通Python函数转为LangChain工具
@tool
def multiply(a: int, b: int) -> int:  # 类型注解：a、b为int，返回值为int（Python语法）
    “””
    multiply two integers
    Args:
        a: First Integer
        b: Second Integer
    “””
    return a * b

# 工具调用（Python字典传参，key对应函数参数名）
print(multiply.invoke({“a”: 2, “b”: 3}))
print(multiply.name)        # 查看工具名称（工具对象属性）
print(multiply.description) # 查看工具描述（来自文档注释）
print(multiply.args)        # 查看工具参数（自动解析）

方案2：Pydantic BaseModel + Field（复杂结构化参数）

Python类继承（继承BaseModel）、Field校验（指定参数描述与约束）

from pydantic import BaseModel, Field
from langchain_core.tools import tool

# Pydantic输入模型（Python类继承，定义参数结构）
class multiplyInput(BaseModel):
    “””this function multiply two number”””
    # Field：指定参数描述，...表示必填（Pydantic语法）
    a: int = Field(..., description=”first arg”)
    b: int = Field(..., description=”second arg”)

# 绑定Schema：通过args_schema参数关联Pydantic模型
@tool(args_schema=multiplyInput)
def multiply(a, b) -> int:  # 无需重复写类型注解，由Schema提供
    return a * b

方案3：Annotated + Field（无额外类，轻量化）

Python类型注解扩展（Annotated），直接为参数添加描述（无需额外定义类）

from typing_extensions import Annotated
from langchain_core.tools import tool
from pydantic import Field

# Annotated：为参数添加类型+描述（Python类型注解扩展）
@tool
def add(
    a: Annotated[int, Field(..., description=”First Arg”)],
    b: Annotated[int, Field(..., description=”Second Arg”)]
) -> int:
    “””add two integer”””
    return a + b

2.3 工具绑定与调用流程

# 工具绑定（Python列表传入工具，bind_tools方法绑定）
tools = [add, multiply]
model_with_tool = model.bind_tools(tools)

# 构造消息（Python列表存储消息）
msg_list = [HumanMessage(“100*20等于多少”)]
ai_msg = model_with_tool.invoke(msg_list)

# 遍历执行工具（Python for循环，遍历tool_calls）
for call in ai_msg.tool_calls:
    # 字典映射，根据工具名称获取对应工具（Python字典取值）
    tool = {“multiply”: multiply}[call[“name”]]
    tool_res = tool.invoke(call)  # 执行工具
    msg_list.append(tool_res)     # 结果加入消息列表

# 最终整合回答
final_res = model_with_tool.invoke(msg_list)

三、结构化输出

通过with_structured_output强制约束输出格式，支持4种方式，适配业务序列化需求。

3.1 Pydantic 嵌套模型

from pydantic import BaseModel, Field
from typing import List, Optional  # Python类型提示，List表示列表，Optional表示可选

# 嵌套Pydantic模型（Python类继承，支持嵌套定义）
class Joke(BaseModel):
    “””给用户讲一个笑话 “””
    setup: str = Field(description=”笑话的开头”)  # str类型，必填
    punchline: str = Field(description=”笑话的笑点”)
    # Optional[int]：可选int类型，默认None
    rating: Optional[int] = Field(default=None, description=”笑话的评分(1~10)”)

class Jokes(BaseModel):
    “””给用户提供的几个笑话”””
    # List[Joke]：列表类型，元素为Joke模型（Python类型提示）
    jokes: List[Joke] = Field(description=”笑话的合集”)

# 绑定结构化输出（传入Pydantic模型，指定输出格式）
model_structured_output = model.with_structured_output(Jokes)
message = [HumanMessage(“分别讲一个关于唱歌和跳舞的笑话”)]
result = model_structured_output.invoke(message)
print(result)  # 直接返回模型对象，可通过属性取值（Python对象属性访问）

3.2 TypedDict 结构化（轻量字典类型约束）

from typing_extensions import TypedDict, Annotated
from pydantic import Field

# TypedDict：轻量字典类型约束（Python字典类型提示）
class Joke(TypedDict):
    # Annotated：为字典键添加描述（无需实例化，直接约束类型）
    setup = Annotated[str, “笑话的开头”]
    punchline = Annotated[str, “笑话的笑点”]
    rating = Annotated[Optional[int], Field(default=None, description=”笑点评分(1~10)”)]

# 绑定结构化输出，include_raw=True返回原始输出
model_structured_output = model.with_structured_output(Joke, include_raw=True)
message = [HumanMessage(“讲一个关于跳舞的笑话”)]
result = model_structured_output.invoke(message)

3.3 JSON Schema 直接定义（自定义JSON格式）

# Python字典定义JSON Schema，指定字段类型、描述、必填项
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”],  # 必填字段（Python列表）
}

# 绑定JSON Schema
model_structured_output = model.with_structured_output(json_schema)
message = [HumanMessage(“讲一个关于跳舞的笑话”)]
result = model_structured_output.invoke(message)

3.4 Union 联合类型（多格式兼容）

from typing import Union  # Python联合类型，支持多种类型兼容

class Standard(BaseModel):
    # Union[Dialog, Joke]：输出可为Dialog或Joke类型（Python联合类型提示）
    output: Annotated[Union[Dialog, Joke], Field(description=”最后输出内容的要求”)]

四、流式传输

4.1 基础流式 stream（同步，迭代器）

chunks = []
# for循环遍历迭代器（Python迭代器语法）
for chunk in model.stream(“讲一个长笑话”):
    chunks.append(chunk)  # 收集所有块（Python列表append方法）
    print(chunk.content, end=””, flush=True)  # 实时打印，flush=True刷新缓冲区

4.2 异步流式 astream（协程，高并发）

import asyncio  # Python异步模块

# 异步函数（async def定义，Python协程语法）
async def async_output():
    # async for：遍历异步迭代器（Python异步迭代语法）
    async for chunk in model.astream(“讲一个言情小故事”):
        print(chunk.content, end=””, flush=True)

asyncio.run(async_output())  # 运行协程（Python异步运行方法）

4.3 自定义流式解析器（生成器yield）

from typing import Iterator  # Python迭代器类型提示

# 自定义解析器（生成器函数，yield关键字生成迭代器，Python生成器语法）
def defined_parser(input: Iterator[str]) -> Iterator[str]:
    buffer = “”
    for chunk in input:
        buffer += chunk
        # 按中文标点切割（Python字符串操作：index找下标、切片）
        while '，' in buffer or '。' in buffer:
            if '，' in buffer:
                stop_index = buffer.index('，')
                yield buffer[:stop_index].strip()  # yield生成每一块内容
                buffer = buffer[stop_index+1:]
            elif '。' in buffer:
                stop_index = buffer.index('。')
                yield buffer[:stop_index].strip()
                buffer = buffer[stop_index+1:]
    yield buffer.strip()  # 生成最后一块内容

# 链式调用自定义解析器
parser = StrOutputParser()
chain = model | parser | defined_parser
for chunk in chain.stream(“写一首关于爱情的诗歌”):
    print(chunk)

4.4 SSE 协议

流式传输需要服务器向客户端主动发送消息。首先我们想到的可能是 WebSocket 协议，这也确实可以，但是需要服务端维护一个长连接，是具有额外的开销。

几乎所有的 LLM 对于流式传输都是用的是 SSE 协议（Server-Sent Event），基于 HTTP 协议，CS 之间建立连接之后，Server 会返回具有 Content-Type: text/event-stream;charset=utf-8 Connection: keep-alive 报头的信息，表示当前为流式传输，客户端不要关闭连接。SSE 协议中，客户端后续不能主动给服务器发送消息，对于这种 LLM 的流式传输的场景，本来就不需要客户端发送第一个请求后，后续继续请求，所以 SSE 更符合使用情景

LangChain 的流式传输，并没有自己封装任何协议，是依赖于大模型供应商提供的流式传输的能力，通过 SSE 协议完成的。其中，AIMessageChunk 是 LangChain 根据大模型供应商的 SSE 接口转换而成的

五、LangChain 核心组件

5.1 消息

• LangChain 中，提供的 SystemMessage、HumanMessage、AIMessage、AIMessageChunk、ToolMessage，都是对不同大模型厂商的接口封装，均继承自 BaseMessage

• BaseMessage 提供了一些通用方法，如 pretty_print，类似 git log --pretty的思想，更清晰美观的展示消息

• LangChain 对话模式

  • 模式一：SystemMessage —> [HumanMessage —> AIMessage] —> [HumanMessage —> AIMessage]… 其中方括号表示一轮对话

  • 模式二：SystemMessage —> [HumanMessage —> AIMessage —> ToolMessage —> AIMessage]

5.2 消息缓存

在使用 LLM 时，我们发现他是能记住一定范围内的上下文的。在 LangChain 中，如果只是使用简单的 invoke 或者 stream 可以发现它并没有记忆：其实 LLM 本身是不具备记忆属性的，每一次调用都是一次全新的推理过程，把上面几轮的用户消息和模型的消息给它，就是作为它的”记忆”，组合拼接成含有”记忆”的答案。

比如下面最简单粗暴的方法，就可以实现记忆的功能

messages = [
    HumanMessage(“我是张三”),
    AIMessage(“你好张三，有什么我可以帮你的吗?”),
    HumanMessage(“我是谁?”)
]
model.invoke(messages).pretty_print();

回答如下，可以发现 LLM 是知道我叫什么的。

================================== Ai Message ==================================
你是张三！😊 很高兴再次见到你。有什么我可以帮你的吗？或者你想聊些什么？

当然这种方法太过简单粗暴了。LangChain 为我们提供了 BaseChatMessageHistory、InMemoryChatMessageHistory 和 RunnableWithMessageHistory，实现上下文的记忆功能。其中，我们使用 invoke 的 config 字段传入 session_id，用来区分不同会话的上下文

cache = {}
def memory_cache(session_id: str) -> BaseChatMessageHistory:
    if session_id not in cache:
        cache[session_id] = InMemoryChatMessageHistory()
    return cache[session_id]
model_with_memory = RunnableWithMessageHistory(model, memory_cache)
config1 = {“configurable”: { “session_id”: “1”}}
model_with_memory.invoke(
    “我叫王家乐”,
    config = config
).pretty_print()
model_with_memory.invoke(
    “我是谁？”,
    config = config
).pretty_print()

不过这种方法现在已经不建议使用了，到了 LangGraph 持久化再往下说

5.3 消息裁剪

一个 LLM 能够处理的上下文是由长度限制的，超过这个限制就需要裁剪，LangChain 允许我们自己设置裁剪策略。

trimmer = trim_messages(
    max_tokens=10, # 最大 token 限制
    token_counter=model, # 通过 语言模型的令牌计数统计
    strategy=”last”, # 保留最后的消息，如果为 “first” 就是保留最新的消息
    allow_partial=False, # 允许消息从中间被裁剪，一般不允许，会导致内容含义改变
    include_system=True, # 是否总是包含第一条 SystemMessage，建议包含，因为其包含对聊天模型的特殊说明
    start_on='human' # 除了第一条 SystemMessage，保留的第一套消息的类型
)

需要注意的是，并不是所有 LLM 都支持通过语言模型的令牌计数来限制 token，比如 qwen-turbo。这里就可以使用另一种方式，通过消息数量来裁剪

trimmer = trim_messages(
    max_tokens=10, # 此时 max_token 表示最大消息数量
    token_counter=len,
    strategy=”last”,
    allow_partial=False,
    include_system=True,
    start_on='human'
)

5.4 消息过滤

除了裁剪，有的时候我们想把所有历史中指定的内容交给 LLM，这时候就需要会话历史裁剪。Langchain 在 langchain_core.messages 中提供了 filter_message 来进行过滤，可以通过消息类型、消息 id 等进行过滤，例如：

messages = [
    SystemMessage(“you are a good assistant”, id='1'),
    HumanMessage(“My name is John”, id='1'),
    AIMessage(“Hello, John”, id='2'),
    HumanMessage(“I'm very happy today”, id='2'),
    AIMessage(“Congratulations! Why?”, id='3'),
    HumanMessage(“Because I hava good grades in my exam! Remember what's my name?”, id='3')
]
filtered_message = filter_messages(messages, include_types=[HumanMessage])
model.invoke(filtered_message).pretty_print()
filtered_message  = filter_messages(messages, include_ids='3') # 这里也可以换成 exclude_...
model.invoke(filtered_message).pretty_print()

输出内容如下，可以看到，第二次发送其实 LLM 并不知道我们叫什么（瞎猜不算）

================================== Ai Message ==================================
I remember your name is John! 🎉 That's wonderful that you got good grades on your exam. I'm happy for you, John! What are you going to do now that you're feeling so great?
================================== Ai Message ==================================
Congratulations! I'm so happy for you! 🎉 Your name is [Your Name], right? (I might need a reminder if you've told me before!) What's your favorite subject?

5.5 消息合并

在历史记录中，可能会出现多个同类型消息连在一起的情况（比如多个 HumanMessage、SystemMessage 连在一起），有些 LLM 不允许这种情况，所以我们可以通过 LangChain 的 merge_message_run 来解决这个问题

messages=[
    SystemMessage(“you are a coding assistant”),
    SystemMessage(“you always feel happy to answer user's questions, but always use Chinese to answer”),
    HumanMessage(“Do you think LangChain is good to use?”),
    HumanMessage(“or do you think I should use LangGraph”),
    AIMessage(“Would you like to tell me, why you wanna using them?”),
    AIMessage(“Then I can give you more infomation?”),
    HumanMessage(“Give me some introduction about them”)
]
merged = merge_message_runs(messages) # 方式一：直接将消息合并
print(model.invoke(messages))

merger = merge_message_runs() # 方式二：构造消息合并器，构建链式调用
chain = merger | model
print(chain.invoke(messages))

5.5 提示词模板

在一个需要批量提出大量类似请求的情境下，为了保证输出的质量和效率，我们可以使用提示词模板。提示词模板可以让我们把精力放在提示词优化上，只需要让应用把相应的变量传给我们即可

5.5.1 字符串模板

from langchain_core.prompts import PromptTemplate
prompt_template = PromptTemplate.from_template(“translate to {language}”) # 直接通过字符串模板初始化
print(prompt_template.invoke(“Chinese”))

prompt_template = PromptTemplate( # 指定变量、模板初始化
    input_variables=[“languages”],
    template=”translate to {languages}”
)
print(prompt_template.invoke(“Chinese”))

5.5.2 消息模板

消息模板在 LangChain 这种直接与聊天模型交互的场景下最为实用。通过指定消息类型和模板的方式，就可以定义一个消息模板。消息模板还实现了 Runnalbe 接口，可以让我们链式调用

消息模板既可以直接 invoke，实例化出模板消息；也可以在流式调用中，通过指定变量值的方式完成调用

chat_prompt_template = ChatPromptTemplate(
    [
        (“system”, “translate the content into {language}”),
        (“user”, “{text}”)
    ]
)
message_template = chat_prompt_template.invoke(
    {
        “language”: “English”,
        “text”: “此曲只应天上有，人间能得几回闻”
    }
)

chain = chat_prompt_template | model
for token in chain.stream({
    “language”: “Chinese”,
    “text”: “This melody should only be found in heaven; how often can it be heard among mortals?”
}):

5.6 从 LangChain_Hub 获取提示词

LangChain_Hub 可以认为是 LangChain 的 GitHub，有许多优质的提示词和模板，我们可以直接像 git 一样拉去下来使用，比如下面使用 prompt_maker 的例子

from langsmith import Client
model = ChatOpenAI(
    model=”qwen-turbo”
)
client = Client()
prompt = client.pull_prompt(“hardkothari/prompt-maker”)
while True:
    task = input(“请输入你的任务（输入 quit 以退出）：\n”)
    if task == 'quit':
        break
    task_prompt = input(“请输入你的任务的提示词，后续会自动优化：\n”)
    chain = prompt | model
    final_prompt = chain.invoke({
        “task”: task,
        “lazy_prompt”: task_prompt
    })
    for token in model.stream([final_prompt]):
        print(token.content, end='')
    print()

5.7 少样本提示

少样本提示就是给 LLM 几个处理好的例子，告诉 LLM 处理的思路、方式，就像初高中常考的新定义问题一样，让 LLM 比着葫芦画瓢

在 LangChain 中，少样本提示通过 FewShotChatMessagePromptTemplate 实现，其中我们可以传入一下内容作为少样本提示：

• example_prompt，是一个 ChatPromptTemplate，用来作为每一个示例的格式化模板
• examples，是一个列表，里面放着我们填入模板的内容

随后，通过消息占位符就可以构造消息列表，传给 LLM，就可以完成有少样本提示的请求

examples = [
    {“input”: “1a1=?”, “output”: “2”},
    {“input”: “1a2=?”, “output”: “3”},
    {“input”: “3a3=?”, “output”: “6”},
]
chat_template = ChatPromptTemplate(
    [
        (“user”, “{input}”),
        (“ai”, “{output}”),
    ]
)

few_shot_template = FewShotChatMessagePromptTemplate(
    example_prompt=chat_template, 
    examples=examples
)
final_prompt = ChatPromptTemplate(
    [
        (“system”, “你是一个数学专家”),
        few_shot_template,
        (“human”, “{input}”)
    ]
)

chain = final_prompt | model
chain.invoke({“input”: “15a10等于多少?”}).pretty_print()

如果使用 PromptTemplate + FewShotPromptTemplate 的话，还可以加上以下关键词（注意新版本 LangChain 不能混用）：

• prefix: 在少样本提示前给予 LLM 的信息

• suffix：在少样本提示后给予 LLM 的信息

5.7.1 推理引导

通过给予 LLM 几个示例问题的思考流程，可以让 LLM 按照我们思考的流程去分析推理，增强结果的可靠度

examples = [
    {
        “question”: “李⽩和杜甫，谁更⻓寿？”,
        “answer”: “””
        是否需要后续问题：是的。
        后续问题：李⽩享年多少岁？
        中间答案：李⽩享年61岁。
        后续问题：杜甫享年多少岁？
        中间答案：杜甫享年58岁。
        所以最终答案是：李⽩
        “””
    },
    {
        “question”: “腾讯的创始⼈什么时候出⽣？”,
        “answer”: “””
        是否需要后续问题：是的。
        后续问题：腾讯的创始⼈是谁？
        中间答案：腾讯由⻢化腾创⽴。
        后续问题：⻢化腾什么时候出⽣？
        中间答案：⻢化腾出⽣于1971年10⽉29⽇。
        所以最终答案是：1971年10⽉29⽇
        “””,
    },
]

chat_template = ChatPromptTemplate.from_messages(
    [
        (“user”, “{question}”),
        (“ai”, “{answer}”)
    ]
)

few_shot_template = FewShotChatMessagePromptTemplate(
    example_prompt=chat_template,
    examples=examples, 
)

final_message = ChatPromptTemplate(
    [
        few_shot_template,
        (“user”, “《星球大战》的导演和《教父》的导演来自同一个国家吗?”)
    ]
)

chain = final_message | model
chain.invoke({}).pretty_print()

5.7.2 使用示例数据增强 LangChain 数据提取能力

前面我们提到过，可以通过 with_structured_output 让 LLM 按照我们的格式输出信息，达到信息提取的效果。这里我们同样可以通过少样本数据提供示例

class Person(BaseModel):
    name: Optional[str] = Field(description=”姓名”)
    skin_color: Optional[str] = Field(description=”这个人的肤色”)
    hair_color: Optional[str] = Field(description=”这个人的发色”)
    height_in_meters: Optional[str] = Field(description=”这个人以米为单位的身高”)
class Data(BaseModel):
    people: List[Person]

examples = [
    (
        “海洋是⼴阔⽽蓝⾊的。它有两万多英尺深。”,
        Data(people=[]), # 没有⼈物信息的情况
    ),
    (
        “⼩强从中国远⾏到美国。”,
        Data(
            people=[
                Person(name=”⼩强”, height_in_meters=None, skin_color=None,
                hair_color=None)
            ]
        ), # 部分信息缺失的情况
    )
]
chat_message = ChatPromptTemplate(
    [
        (“system”, “你是一个任务信息提取专家，如果给你的描述中，没有相关信息，则相应字段为 None”),
        (“placeholder”, “{example_messages}”),
        (“user”, “{new_message}”)
    ]
)

example_messages = []
for txt, tool_call in examples:
    if tool_call.people:
        ai_response = “未识别到人”
    else:
        ai_response = “识别到人”
    example_messages.extend(
        tool_example_to_messages(
            txt, [tool_call], ai_response = ai_response
        )
    )

model_with_structured_output = model.with_structured_output(Data)
chain = chat_message | model_with_structured_output
print(chain.invoke({“example_messages”: example_messages, “new_message”: “篮球场上，⾝⾼两⽶的中锋王伟默契地将球传给⼀⽶七的后卫挚友李明，完成⼀记绝杀。” “这对⽼友⽤⼗年配合弥补了⾝⾼的差距。”}))

5.8 示例选择器

一方面 LLM 的上下文窗口是由最大限制的，这意味着我们不能无脑的把各种示例都丢给 LLM；另一方面，当示例过多时，LLM 会出现混淆示例的情况，反而起到反面作用。

所以我们要对示例有所选择。

5.8.1 通过长度选择器选择

使用 LengthBasedExampleSelector，通过指定 max_length 的最大片段个数，分割示例。默认片段由空格、换行符、制表符这些空白字符分割

使用了示例选择器的少样本提示，不需要再使用 examples 字段，由 example_selector 提供选择过后的样本

examples=[
    {“input”: “happy”, “output”: “sad”},
    {“input”: “tall”, “output”: “short”},
    {“input”: “energetic”, “output”: “lethargic”},
]
prompt_template = PromptTemplate(
    input_variables=[“input”,  ”output”],
    template=”input: {input}\noutput: {output}”
)
length_selector = LengthBasedExampleSelector(
    examples=examples,
    example_prompt=prompt_template,
    max_length=25 # 按空格、换行、制表符分割之后的片段的数量
)
few_shot_template = FewShotPromptTemplate(
    example_selector=semantic_selector,    
    example_prompt=prompt_template,
    prefix=”给出每个输入的反义词”,
    suffix=”input: {adjective}\n”,
    input_variables=[“adjective”]
)
print(few_shot_template.invoke({“adjective”: “vim”}).to_messages()[0].content)

5.8.2 通过嵌入模型进行语义选择

来看下面的两个例子

1. 苹果很甜
2. 苹果有自己的笔记本电脑

从语义上说，这两个是完全不同的，一个是水果，一个是公司。通过嵌入模型对我们的若干个例子转化成向量进行分析，就可以找到相似点、不同点，从而提炼成我们需要的数目的例子

在 LangChain 中，使用 语义示例选择器，我们可以使用 SemanticSimilarityExampleSelector

• OpenAIEmbeddings，用来生成度量语义向量的嵌入模型
• Chroma，VectorStore，用来存放和管理向量的数据库
• k，指定最后选择剩下的示例的个数

examples=[
    {“input”: “happy”, “output”: “sad”},
    {“input”: “tall”, “output”: “short”},
    {“input”: “energetic”, “output”: “lethargic”},
]
prompt_template = PromptTemplate(
    input_variables=[“input”,  ”output”],
    template=”input: {input}\noutput: {output}”
)

semantic_selector = SemanticSimilarityExampleSelector.from_examples(
    examples, OpenAIEmbeddings(model=”text-embedding-v3”), Chroma, 2)
    
few_shot_template = FewShotPromptTemplate(
    example_selector=semantic_selector,    
    example_prompt=prompt_template,
    prefix=”给出每个输入的反义词”,
    suffix=”input: {adjective}\n”,
    input_variables=[“adjective”]
)

5.8.3 MMR

MMR 即最大边际相关性，Maximum Margin Relavance，通过将冗余度和相关性结合共同为样本打分的方式，使得样本既能够保证主题的相关性，又能够保证样本的多样性。

举个例子，在招聘的时候，最大语义相关性就是给每个人直接都打一个分，直接选最高的；最大边际相关性就是，先选出一个分最高的（最相关的），后续每选择下一个人的时候，既看分数高低（相关性），有看这个人和其他已经选出的人的技能重复性，最后选出来一个每个人的技能都相对独立的团队

MMR 可以用在信息推荐，保证主题是符合用户喜爱的主题的，同时形式等尽可能多样，缓解信息茧房

在 LangChain 中，为我们提供了 MaxMarginalRelevanceExampleSelector 来进行最大边际相关性的信息筛选，使用格式与最大语义相关性的相同

5.8.4 语义 NGram 重叠

首先我们需要明确一些概念

5.8.4.1 什么是 NGram？

• NGram 是指字符序列中连续的词或字符
• 比如 I Like Apple，I 、Like、Apple 都可以是 NGram

5.8.4.2 什么是 NGram 重叠

• 两个字符序列中相同的 NGram 就是 NGram 重叠
• 比如句子 Apple is red 和 I have a red apple 中，apple、red 就是重叠的 NGram，NGram 重叠度越高，代表两个文本序列在用词上越相近

5.8.4.3 什么是语义 NGram 重叠

• 上述的比较过程中，不直接比较词本身是否相同，而是比较词在向量空间中的相似性。
• 比如，句子 I love you 和 I like you 的语义 NGram 重叠，就要比 I hate you 的要高
• NGram 在检测抄袭上有很大用处，只是通过重新表述、同义词替换得来的文本之间就会有很大的语义 NGram 重叠度

5.8.4.4 LangChain 中使用 NGramOverlapExampleSelector

在 LangChain 中，想要使用通过语义 NGram 筛选的示例选择器，需要使用 langchain_community 包中的 NGramOverlapExampleSelector

NGramOverlapExampleSelector 的参数使用基本相同，但是多了一个 threshold 参数，用来表示筛选的门槛，一般取值在0.0 ～ 1.0。其中：

• -1.0 表示保留所有示例，仅仅通过语义 NGram 重叠度进行降序排序
• 0.0 表示只删除掉毫不相关的示例，其他只要有一点相关的都保留
• 1.0 表示完全相关的保留，其他删除，但一般很难做到

• 1.0 表示全部删除，提供一个空示例集

在使用时，需要导入包 nltk，这个包在各种自然语言处理的包中都很常用

5.9 输出解析器 (Output Parser)

5.9.1 输出解析器和 with_return_object 的区别

• 输出解析器是 LangChain 为我们提供的一个功能，with_return_object 是大模型的功能，使用后返回了一个带格式化返回结果的大模型
• 使用上，输出解析器可以进行链式调用，而 with_return_object 只能通过返回的具有 Runnable 接口的对象使用

5.9.2 StrOutputParser

前面已经提到过了，这里省略

5.9.3 PydanticOutputParser

这里我们可以定义一个 Pydantic 类，通过 get_format_instruction 方法，获得一个给大模型的、不影响大模型推理思考、只指导大模型在返回结果时按 Pydantic 类要求的结构返回的指令，同时，因为这个指令是固定的，所以还可以通过 PromptTemplate 的 partial_variables 字段固定下来，不用每次传入

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser, PydanticOutputParser
from pydantic import BaseModel, Field
from typing import Optional
from langchain_core.prompts import PromptTemplate

model = ChatOpenAI(
    model = “qwen-turbo”
)
def pydantic_output_parser_test():
    class Joke(BaseModel):
        setup: Optional[str] = Field(description=”笑话的开头”),
        punchline: Optional[str] = Field(description=”笑话的妙语”),
        rate: Optional[int] = Field(default=None, description=”笑话的评分(1~10)”)

    pydantic_output_parser = PydanticOutputParser(pydantic_object=Joke)
    prompt_template = PromptTemplate(
        template=”回答用户指令：{format_instruction} \n {query}\n”,
        input_variables=[“query”],
        partial_variables={“format_instruction”: pydantic_output_parser.get_format_instructions()}
    )
    chain=prompt_template | model | pydantic_output_parser
    print(chain.invoke({“query”: “将一个关于电脑的笑话”}))

5.9.4 JsonOutputParser

在 PromptTemplate 中的使用和 JsonOutputParser 一样，可以不传结构化要求，只是将正文按照 json 的格式返回；也可以通过指定 pydantic_object 的方式，让返回结构按照 pydantic 类的结构返回 json 格式

def json_output_parser_test():
    output_parser = JsonOutputParser()
    prompty_template = PromptTemplate(
        template=”answer user's question: {input}\n {format_instruction}”,
        input_variables=[“input”],
        partial_variables={“format_instruction”: output_parser.get_format_instructions()}
    )
    chain=prompty_template|model|output_parser
    print(chain.invoke({“input”: “give me a joke about computer”}))

5.10 RAG (Retrieval-Augmented Generation)

5.10.1 RAG 的流程

LLM 最擅长推理，但是不擅长进行信息的获取，因为模型的训练时间是有截止日期的。所以第一反应是让大模型和搜索引擎结合起来，但是这只能获得最新的有限信息，比如公司内部文档等都无法指通过搜索引擎获得。

检索增强生成，大体分为两个阶段 离线数据处理 和 在线检索，将私有数据或者浏览器检索获得最新知识整理获得知识库

离线数据处理 更详细来说，分为下面的三个部分

1. 文档加载，将各种形式的文档加载（拷贝）过来，形式可以是 PDF、MarkDown、Python、Java、C++ 代码等
2. 文档分割，将加载完成的文档，切分成若干的文档块。在 LangChain 中，由 Document 对象保管文本内容 (page_content) 和属性 (metadata)
3. 存储，首先把分割出的文档块转换成向量的形式，这个过程称为嵌入；其次将这些向量存储管理起来供给下面的流程使用，比如使用 LangChain_Chroma 提供的向量数据库
   在线检索，是将上面处理好的向量数据库筛选出来符合主题的部分，交给 LLM 结合问题，整合出更可靠的答案。更详细的，分为这两个部分
4. 检索，通过语义相关性，筛选出知识库中和问题相关的 (MSR，MMR等) 知识
5. 输出，结合筛选出来的资料，和问题本身，通过 LLM 的推理，生成答案输出

5.10.1.2 PyPDFLoader

langchain_community 提供的 PDF 文档加载器，只需要通过文档路径就可以构建 PyPDFLoader，我们可以通过其获得文档原始内容、页数、结构化的属性信息等

from langchain_community.document_loaders import PyPDFLoader
pdf_loader = PyPDFLoader(“/home/human/Document/bite/MySQL/test.pdf”)
doc = pdf_loader.load()
print(f”pdf has {len(doc)} pages”)
print(f”the first page's first 200 words: {doc[0].page_content[:200]}”)
print(f”the first page's metadata {doc[0].metadata}”)

5.10.1.3 UnstructuredMarkdownReader

{'Image', 'Title', 'ListItem', 'Table', 'NarrativeText', 'UncategorizedText'}

langchain_community 提供的 MarkDown 文档加载器，依旧只需要文档路径就可以构建，但是多了一个 mode 参数，可以提供两种选项

1. single，将文件原始内容和属性都放在一个 Document 对象中返回，不进行类型的切分

2. elements，将文件原始内容和属性放在多个 Document 对象中返回，其中 metadata 中的 catagory 会对文本的 Markdown 语法角色进行划分，具体包括下面的几种属性

   1. Image，Markdown 中插入的图片，通过 !(comment)[path_to_file] 的语法
   2. Title，# 语法的标题，其中，各个级别的标题之间，通过 metadata 中的 parent_id 属性确定，我们的每个元素都有自己的 element_id
   3. ListItem，包括有序列表和无需列表
   4. Table，使用 Markdown 语法创建的表格
   5. NarrativeText，叙述性文本，一个或者多个连续的段落
   6. UncategorizedText，未分类文本，主要是各种脚注、注释等小文本

5.10.2 文档分割

在通过各种形式的文档完成文档加载之后，并不能直接录入知识库。一方面，大模型的上下文窗口有限，不分割可能压根就塞不进去；另一方面，分割之后，有利于进行管理，更细的粒度也可以让语义搜索更加精确

文档分割也有几种形式，可以根据文档长度，也可以根据文档语义。前者大体上分为 根据字符分割 或者 根据Token分割

5.10.2.1 通过字符数进行分割

通过 LanghChain 提供的 CharacterTextSpliter 进行分割，根据指定的字符数进行分割，同时也需要设置每个小块之间有一定的重叠，否则可能会造成语义割裂或者上下文缺失（比如”兽人永不为奴，除非保持包吃包住”这句话）。这两项通过 chunk_size 和 chunk_overlap 指定。

关于分割符，默认是 \n\n ，并且会按照 “\n\n” “\n” “ “ “” 的顺序依次进行尝试。原因是，切分除了把文档化成小块之外，还有一个原则就是尽可能不要破坏原来的语义。所以当无法在既符合长度又保证语义的情况下，就会选择把信息保留下来，并通过警告的方式告知。（把 chunk_size 设置为200左右可以减少超过的分割）

可以通过 length_function 字段指定分割长度的计算方式

5.10.1.2 通过 token 数进行分割

依然是使用 CharacterTextSpliter，但是使用 from_tiktoken_encoder 来指定根据 tiktoken 分词器进行分割的文档分割器。其中，可以指定 encoding_name，可以指定为 cl100k_base，这种编码方式对于 OpenAI 的模型更加准确。

通过 token 数进行分割时，chunk_size chunk_overlap 的单位都变成 token 数量。为了不破坏语义，依然会保留无法切割的部分，即使超过 chunk_size

char_spliter = CharacterTextSplitter.from_tiktoken_encoder(  
    chunk_size=100,  
    chunk_overlap=20,  
    encoding_name=”cl100k_base”  
)  
for chunk in char_spliter.split_documents(doc)[:10]:  
    print(chunk.page_content)  

5.10.1.3 RecursiveCharacterSpliter

强制按照指定的 chunk_size 进行分割，即使会造成语义的破坏，其他使用方式和前面两种一样。

除此之外，在进行中文的分割时，可能会将一个词分成两个单独的字，建议使用 separators 进行分隔符的指定

recursive_spliter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(  
	chunk_size=100,  
	chunk_overlap=20,  
	encoding_name=”cl100k_base”,  
	separators=[“\n\n”, “\n”, “，”, “。”, “ “, “”] 
)

5.10.3 文本向量

计算机本身是无法理解人类的自然语言、图像、音频的，但是将他们转换为具有若干维度的向量，就可以让计算机解析出其中的语义进行推导、生成。我们可以通过嵌入模型，完成文本到向量的转换。向量具有的维数阅读，质量就越高，语义也就越精细，更有利于理解。比如使用 Ollama 本地部署的最丐版的嵌入模型生成的向量的维数就是 384

在向量空间中，对于语义的处理，主要有两种：

• 欧式长度：即我们高中所学的两点之间的距离，距离越近相似度越高
• 余弦相似度：只关注两个向量在方向上的差异，差异越小，语义上的差别就越小

余弦相似度在度量语义时，使用的更多一下。使用文本向量来分析语义，在语义检索、检索增强生成、风险预警、推荐系统等都有用处

5.10.3.1 在 LangChain 中使用嵌入模型

可以使用ChatGPT的嵌入模型来进行文本的嵌入，我这里使用的是 Ollama 进行本地部署，最丐版的 all-minilm，只有 45MB

embeddings = OllamaEmbeddings(  
    model=”all-minilm”  
)  
texts = [chunk.page_content for chunk in spliter.split_documents(doc)]  # doc 是切分好的文档块
  
vectors = embeddings.embed_documents(texts)  # 嵌入文档
print(f”第一个文档共有{len(vectors[0])}个向量”)  
  
query_result = embeddings.embed_query(“MMR”)  # 嵌入单个查询
print(f”向量的前五个维度{query_result[:5]}”)

5.10.3 向量存储

向量数据库存在的意义，就是把我们向上面这样得到的文本向量存储起来，并进行高效的管理，使我们能够高效的进行文本向量的检索

5.10.3.1 向量索引

像传统的关系型数据库一样，向量数据库也需要建立索引。但是向量数据库更多是为了解决 MySQL 这种传统数据库所不擅长的模糊匹配操作。精确的找到我们需要的向量当然是最好的，但是向量的维数、向量的个数使得这件事变得很困难，最终导致 维度灾难

所以向量数据库采取了牺牲精度答复换取效率的做法。比如最常被使用的方法 ANN 近似最相邻搜索

分层导航小世界有很多种实现方法，比如哈希桶，不过更常用的是 HNSW 分层导航小世界，采取局部贪心的策略（通过向量余弦相似度、欧式长度等方法贪心），顶部的层稀疏，底部的层稠密，实现接近 O(logN) 的时间复杂度的检索

5.10.3.2 内存向量数据库

向量存储

可以使用 langchain_core 包中的 InMemoryStore，将分割好的文本块直接交给指定过嵌入模型的向量数据库即可，随后返回一个 id 列表，代表为每一个文本块生成的向量。

向量查询

通过 get_by_ids 进行，传入向量 id 列表，返回 Document 对象列表

向量删除

通过 delete 进行，传入向量 id （注意，get_by_ids 和 delete 都需要传入列表，不能直接传入向量 id 字符串，并且 delete 还要通过关键字传参）

向量的语义查询

语义的查询可以通过多种方式，比如前面提到的欧式距离和余弦相似度。InMemoryOrderSearch 是使用的余弦相似度，使用其 similarity_search 方法

def _vector_filter(candidate: Document) -> bool:  
    return candidate.metadata.get("source") != "xxx"  
  
ret = vector_store.similarity_search(  # 返回的是一个 Document 对象列表
    query="Runnable",  
    k = 2,  
    filter=_vector_filter  # 可以定义一个输入参数是 Document 对象，返回bool的自定义过滤条件函数，返回 true 表示符合条件保留
)  

截至目前，RAG 离线数据处理（文本加载、文本分割、文本嵌入）和 在线检索（检索 + 生成输出）就基本完成了

5.10.3.3 Redis向量存储数据库

在 RediSearch 中，采取 Index + IndexField + RedisHash 的方式为我们提供快速的索引，类似于 MySQL 的 表 + 列 + 值 的方式。查询时，首先根据 index_name 明确去哪个 Index 下面查询，之后通过 IndexField 中各种索引条件，根据查询获取到对应的文档 id，最后通过获取到的文档 id 到实际存储文件内容的 Redis Hash 中查询

Index 并不负责存储数据，只是对所有的 IndexField 进行管理；IndexField 存储着所有我们用来查询的索引，包括自带的（如 embeddings）以及我们通过 schema 指定的。在 IndexField 中，所有索引都有自己的类型，比如：

• TEXT，全文搜索的字符串，支持分词搜索、模糊匹配
• TAG，精确匹配的标签
• NUMERIC，整数/浮点数，支持范围查询、统计等

Redis 数据库的初始化：index_name 指定的是 Index 的名称，作为第一次筛选的条件；redis_url 指定的是 redis 服务器；metadata_schema 指定的是自定义的 IndexField 索引字段

config = RedisConfig(
    index_name = "QA",
    redis_url = "redis://localhost:6379",
    metadata_schema=[
        {"name": "catagory", "type": "tag"},
        {"name": "num", "type": "numeric"}
    ]
)
redis_store = RedisVectorStore(
    embeddings=embedding,
    config=config
)

为分割出的文档添加自定义索引信息，并将小文本及其信息传入 Redis 向量数据库

for i, document in enumerate(splited_docs, start=1):
    document.metadata["catagory"] = "QA"
    document.metadata["num"] = i
ids = redis_store.add_documents(splited_docs)

与 InMemoryVectorStore 一样，RediSearch 也支持 similarity_search 以及添加过滤条件，同时还可以显示匹配分数（通过 similarity_search_with_score，在 page_content metadata 之外又加上了 score 字段，越小表示匹配度越高）

MMR

在 RediSearch 中，支持通过 MMR 的方式对向量进行查询，通过 max_marginal_relevance_search 进行，

mmr_results = vector_store.max_marginal_relevance_search( 
	query="数据库表怎么设计的？", 
	k=2, 
	fetch_k=10, 
	filter=filter_condition 
)

• fetch_k 是初步筛选出的最相关的 fetch_k 个文档，作为初始的向量池
• 筛选出最初的向量池后，从这个向量池中，选出 k 个与其他不太相似的文档作为最终结果返回，使得结果既相关又多样化

检索器

使用向量数据库作为检索器，根据用户输入检索，非常适合替代 LIKE 类的查询。RediSearch 直接为我们提供了 as_retriever 方法，为我们返回一个 Runnable 对象，直接传入查询文本即可完成查询

as_retriever 还为我们提供了如下参数：

• search_type，搜索方式，默认是 similarity ，也可以指定为 mmr 以及 similarity_score_threshold
• search_kwargs：
  • k，结果的最大数量
  • fetch_k，给 MMR 的初始向量池大小

检索器虽然是 Runnable 对象，但是不提供流式操作，调用 stream 接口也是一次返回所有内容。

检索器也可以通过 @chain 定义，通过一个输入 str，输出 List[Document] 的函数，即可完成（这里真正的检索器是 Runnable 对象，我们定义的只是一个方便我们使用的具有检索器特点的函数

@chain
def chain_retriever(input: str) -> List[Document]:
    return redis_store.similarity_search(input, k=1)
ret = chain_retriever.invoke("LangChain的流式传输")