1. PromptTemplate 与 ChatPromptTemplate 的区别

1.1 PromptTemplate – 单字符串输出

PromptTemplate 用于传统的文本补全模型（如 text-davinci-003），输出一个纯字符串。

from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate.from_template("请写一首关于{topic}的诗。")
result = prompt.format(topic="边塞")
print(result)
# 输出：请写一首关于边塞的诗。

1.2.ChatPromptTemlate – 消息列表输出

ChatPromptTemplate 用于聊天模型（如 GPT-3.5/4、DeepSeek Chat），输出一个 List[BaseMessage]，每条消息包含 role 和 content。

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位边塞诗人。"),
    ("user", "请写一首关于{topic}的诗。")
])
messages = prompt.format_messages(topic="边塞")
print(messages)
# 输出：[SystemMessage(content='你是一位边塞诗人。'), HumanMessage(content='请写一首关于边塞的诗。')]

1.3 MessagesPlaceholder – 动态插入历史对话

ChatPromptTemplate 特有的 MessagesPlaceholder 可以动态插入历史对话记录，这是 PromptTemplate 无法直接做到的（除非你手动把历史转成字符串）。

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是助手"),
    MessagesPlaceholder("history"),
    ("user", "{input}")
])

history_data = [
    ("user", "请给我写一首李白的唐诗"),
    ("assistant", "床前明月光，疑似地上霜。举头望明月，低头思故乡"),
    ("user", "好诗再来一首"),
    ("assistant", "窗含西岭千秋雪，门泊东吴万里船"),
]

# 正确调用
messages = prompt.format_messages(history=history_data, input="你的提问")
print(messages)

1.4 对比表格

特性	PromptTemplate	ChatPromptTemplate
适用模型	传统的 文本补全模型（如 GPT-3、text-davinci-003）	聊天模型（如 GPT-3.5/4、DeepSeek Chat、文心一言）
输出内容	一个 单一的字符串（纯文本 prompt）	一个 消息列表（List[BaseMessage]），每条消息有 role（如 system、user、assistant）和 content
输入变量	简单字符串格式化（如 {topic}）	支持结构化历史消息占位符（如 MessagesPlaceholder）
典型使用场景	生成文章、翻译、代码补全等单轮文本生成	多轮对话、带系统指令的聊天、对话记忆管理

2. 基础链式调用

LangChain 允许通过 | 运算符将多个组件串联成一条“链”（Runnable 序列），链中的每个组件都必须实现 Runnable 接口。

2.1 链式调用的核心概念

• 使用 | 连接各个组件，形成 RunnableSerializable 对象。
• 调用链的 invoke 方法获得最终结果，或调用 stream 获得流式输出。
• 典型结构：提示词模板 | 模型 | 输出解析器

2.2 最简单的链：提示词模板 + 模型

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_deepseek import ChatDeepSeek

# 定义提示词模板
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位边塞诗人，可以作诗。"),
    MessagesPlaceholder("history"),
    ("user", "请给我写一首孟浩然的唐诗"),
])

# 历史消息（可以动态传入）
history_data = [
    ("user", "请给我写一首李白的唐诗"),
    ("assistant", "床前明月光，疑似地上霜。举头望明月，低头思故乡"),
    ("user", "好诗再来一首"),
    ("assistant", "窗含西岭千秋雪，门泊东吴万里船"),
]

# 初始化模型
model = ChatDeepSeek(
    model_name="deepseek-chat",
    api_key="your-api-key",      # 建议从环境变量读取
    api_base="https://api.deepseek.com/v1",
    temperature=0.5,
    max_tokens=1024,
    top_p=1
)

# 构建链
chain = chat_prompt | model

# 执行链
result = chain.invoke({"history": history_data})
print(result.content)   # AIMessage 对象，用 .content 获取文本

2.3 输出解析器：StrOutputParser

模型返回的是 AIMessage 对象，若只需纯文本内容，可以添加 StrOutputParser 自动提取 .content。

from langchain_core.output_parsers import StrOutputParser

chain = chat_prompt | model | StrOutputParser()
result = await chain.invoke({"history": history_data})
print(result)  # 直接输出字符串

2.4 完整 FastAPI 示例

以下是一个结合 FastAPI 的异步接口示例，注意 实际生产环境建议使用 ainvoke 以避免阻塞事件循环

from fastapi import FastAPI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_deepseek import ChatDeepSeek
import config_data as config

app = FastAPI()

# 全局定义提示词模板（避免每次请求重新创建）
 prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位边塞诗人，可以作诗。"),
    MessagesPlaceholder("history"),
    ("user", "请给我写一首孟浩然的唐诗"),
])

# 全局模型实例
 model = ChatDeepSeek(
    model_name=config.DEEPSEEK_MODEL,
    api_key=config.DEEPSEEK_API_KEY,
    api_base=config.DEEPSEEK_API_URL,
    temperature=0.5,
    max_tokens=1024,
    top_p=1
)

# 链（包含输出解析器）
base_chain = prompt | model | StrOutputParser()

@app.post("/chat")
async def chat():
    # 历史数据在实际应用中应从请求参数或会话存储中获取
    history_data = [
        ("user", "请给我写一首李白的唐诗"),
        ("assistant", "床前明月光，疑似地上霜。举头望明月，低头思故乡"),
        ("user", "好诗再来一首"),
        ("assistant", "窗含西岭千秋雪，门泊东吴万里船"),
    ]
    # 调用链（注意：invoke 是同步方法，在 async 函数中建议使用 ainvoke）
    result =await base_chain.invoke({"history": history_data})
    return {"message": result}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main2:app", host="127.0.0.1", port=8001, reload=True)

3. JsonOutputParser：将模型输出解析为字典

3.1 基本概念

JsonOutputParser 是 LangChain 内置的输出解析器，用于将模型返回的 JSON 字符串解析为 Python 字典（dict）。它要求模型输出的内容是一个可被 json.loads() 正确解析的合法 JSON 字符串。

• 输入：AIMessage（模型的原始输出）
• 输出：dict（解析后的 JSON 对象）

使用场景：结构化数据提取、API 参数生成、多步链中的中间结果传递等。

3.2 基础用法

from fastapi import FastAPI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.output_parsers import StrOutputParser
from langchain_deepseek import ChatDeepSeek
import config_data as config
app = FastAPI()
str_parser=StrOutputParser()
json_parser=JsonOutputParser()
@app.post("/chat")
async def chat():
    first_prompt=PromptTemplate.from_template(
        "我邻居姓：{lastname}，刚刚出生了{gender},请起名字，并封装到JSON格式返回，"
        "请将结果封装为JSON格式,要求key是name,value就是你起的名字，请严格遵守格式"
    ) 
    second_prompt = PromptTemplate.from_template("姓名{name}，请帮我解析含义")
    model = ChatDeepSeek(
        model_name=config.DEEPSEEK_MODEL,        # 例如 "deepseek-chat"
        api_key=config.DEEPSEEK_API_KEY,    # 注意是 api_key，不是 dashscope_api_key
        api_base=config.DEEPSEEK_API_URL, 
        temperature=0.5,
        max_tokens=1024,
        top_p=1
    )
  
    chain =first_prompt | model | json_parser|second_prompt |model | str_parser
    result = chain.invoke({"lastname":"张","gender":"男"})
    return {"message":result}
    
   
if __name__ == '__main__':
    import uvicorn
    uvicorn.run("main2:app",host="127.0.0.1",port=8001,
                reload=True) 

4. 多模型链式调用与 RunnableLambda

4.1 多模型链的概念

LangChain 支持将多个模型调用串联成一个整体链。典型场景：

1. 第一次模型调用：根据输入生成一些结构化数据（如 JSON）。
2. 中间处理：解析第一次的输出，提取关键字段。
3. 第二次模型调用：基于提取的字段继续执行新任务（如分析、扩展、翻译等）。

这种模式可以构建更复杂的应用，例如“起名 + 解析名字含义”、“信息抽取 + 文案生成”等。

4.2 链的基本结构

chain = first_prompt | model | parser_or_lambda | second_prompt | model | str_parser

• first_prompt：第一次调用使用的提示词模板。
• model：语言模型（可以是同一个实例，也可以是不同模型）。
• parser_or_lambda：将第一个模型的 AIMessage 输出转换为第二个提示词所需的 dict 参数。
• second_prompt：第二次调用使用的提示词模板。
• str_parser：最终将第二次模型输出的 AIMessage 转为字符串。

4.3 RunnableLambda：将任意函数转换为 Runnable

RunnableLambda 是 LangChain 提供的适配器，可以将普通函数或 lambda 表达式包装成 Runnable 对象，从而无缝插入链中。

4.3.1 语法与用法

from langchain_core.runnables import RunnableLambda

 my_fun=RunnableLambda(lambda ai_msg:{"name": json.loads(ai_msg.content.strip()["name"])})

输入与输出：

• 输入：上游组件传来的任何对象（通常是 AIMessage、dict 或 str）。
• 输出：函数返回的值（可以是 dict、str 或任意可被下游提示词模板接受的对象）。

4.4 完整示例：两次模型调用 – “起名 + 解析含义”

场景描述

• 第一次调用：根据姓氏和性别生成一个名字，并以 JSON 格式返回（{“name”: “某某”}）。
• 提取 JSON 中的 name 字段。
• 第二次调用：解析这个名字的含义。

from fastapi import FastAPI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_deepseek import ChatDeepSeek
from langchain_core.runnables import RunnableLambda

import config_data as config
import json
app = FastAPI()

# 最终输出解析器
str_parser = StrOutputParser()

# 初始化模型（共用同一实例）
model = ChatDeepSeek(
    model_name=config.DEEPSEEK_MODEL,
    api_key=config.DEEPSEEK_API_KEY,
    api_base=config.DEEPSEEK_API_URL,
    temperature=0.5,
    max_tokens=1024,
    top_p=1
)

@app.post("/chat")
async def chat():
    # 第一步：起名提示词（要求输出 JSON）
    first_prompt = PromptTemplate.from_template(
        "我邻居姓：{lastname}，刚刚出生了{gender}，请起一个名字。"
        "请将结果封装为JSON格式，要求key是name，value就是你起的名字，严格遵守格式，只输出JSON。"
    )
    
    # 第二步：解析含义提示词（使用第一次解析出的 name）
    second_prompt = PromptTemplate.from_template( "请解析名字“{name}”的含义，用中文简要说明。" )
    
    my_fun=RunnableLambda(lambda ai_msg:{"name": json.loads(ai_msg.content.strip()["name"])})
    chain = ( first_prompt| model |  my_fun| second_prompt | model| str_parser )
    result = chain.invoke({"lastname":"张","gender":"男"})
    return {"message":result}
    
    
if __name__ == '__main__':
    import uvicorn
    uvicorn.run("main2:app",host="127.0.0.1",port=8001,
                reload=True)  
    

5. 向量数据库

5.1 Chroma

5.1.1 什么是Chroma？

chroma是一个开源的数据向量库，专门用于存储和检索向量化的数据(embeddings）。它让计算机能够“理解”文本含义，进行智能搜索，是构建 RAG（检索增强生成）等 AI 应用的重要基础设施。
什么是Embedding 向量化基础
定义：将文本转换为数字向量（如 768、384、1024 维），表达语义属性。
维度选择：取决于模型设计、参数量、训练数据。简单模型维度低（100_{300），复杂模型维度高（768}1536），可捕捉细粒度语义（医疗、法律）或跨模态（图文）。

相似度计算：常用余弦相似度（点积/模长乘积），值越大越相似。例如：“跑步鞋”与“运动鞋”语义相近。

示例模型：DashScopeEmbeddings（text-embedding-v1，输出 1536 维）、SentenceTransformer。

5.1.2 核心功能

**存储向量**：保存文本转化为向量标识
**语义检索**：根据含义而非关键词查找相似内容
**持久化**：向量数据可以保存到磁盘，程序启动后继续使用
**元数据管理**：处理向量，还可以存储相关的附加信息，可以根据这些信息进行元素据过滤

5.1.3 典型应用

智能文档搜索（客服知识库、法律文书检索）
智能文案推荐
相似商品推荐
问答系统（RAG）

5.1.4 嵌入模型选择

本地运行：HuggingFaceEmbeddings（需下载模型）
云端调用：DashScopeEmbeddings（阿里灵积）、OpenAIEmbeddings 等

5.1.5 安装与准备

---

安装

pip install langchain-chroma

langchain-chroma 是 LangChain 对 Chroma 的集成包，chromadb 是 Chroma 核心库。

5.1.6 基本用法(代码示例）

5.1.6.1 初始化嵌入模型（以阿里云DashSope为例）

from langchain_community.embeddings import DashScopeEmbeddings

embedding = DashScopeEmbeddings(
    model="text-embedding-v1",
    dashscope_api_key="your-dashscope-api-key"   # 替换为真实 key
)

若使用本地模型，可替换为 HuggingFaceEmbeddings(model_name=“sentence-transformers/all-MiniLM-L6-v2”)

5.1.6.2创建向量库（支持持久化）

from langchain_chroma import Chroma

vector_store = Chroma(
    collection_name="my_knowledge",
    embedding_function=embedding,
    persist_directory="./chroma_db"   # 数据存储目录，不设置则仅内存
)

5.1.6.3添加文档（Document 对象）

from langchain_core.documents import Document

doc_list = [
    Document(page_content="苹果是一种水果", metadata={"category": "food"}),
    Document(page_content="iPhone是苹果公司的产品", metadata={"category": "tech"}),
    Document(page_content="牛顿被苹果砸中，发现了万有引力", metadata={"category": "history"}),
]

vector_store.add_documents(doc_list)

5.1.6.4 直接添加文本（add_texts）

from datetime import datetime

texts = ["春眠不觉晓，处处闻啼鸟"]
metadatas = [
    {
        "source": "唐诗三百首",
        "create_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "operator": "admin",
        "user_id": "1001"
    }
]

vector_store.add_texts(texts=texts, metadatas=metadatas)
metadatas 长度必须与 texts 一致，若不需要元数据可传 None。

5.1.6.5 相似度检索（返回文档）

query = "水果"
results = vector_store.similarity_search(query, k=2)

for doc in results:
    print(f"内容：{doc.page_content}\n元数据：{doc.metadata}\n")

5.1.6.6 带分数的相似度检索

docs_with_score = vector_store.similarity_search_with_score(query, k=2)

for doc, score in docs_with_score:
    print(f"分数：{score:.4f} | 内容：{doc.page_content}")

分数默认为 L2 距离（越小越相似）。若使用余弦距离，分数为 1 - cosine_similarity，同样越小越相似。

5.6.7 元数据过滤检索

results = vector_store.similarity_search(
    query="苹果",
    k=1,
    filter={"category": "tech"}   # 仅返回 category 为 tech 的文档
)

5.1.6.8 MMR 检索器（兼顾相关性与多样性）

retriever = vector_store.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 2,            # 最终返回的文档数
        "fetch_k": 5,      # 初始检索的候选文档数
        "lambda_mult": 0.5 # 0=完全多样性，1=完全相关性
    }
)

mmr_results = retriever.invoke("苹果")

for doc in mmr_results:
    print(doc.page_content)

5.1.6.9 重新加载已有的向量库

# 重新启动程序后，只需使用相同参数初始化即可自动加载
loaded_store = Chroma(
    collection_name="my_knowledge",
    embedding_function=embedding,
    persist_directory="./chroma_db"
)

# 无需再调用 add_documents，直接检索
loaded_store.similarity_search("水果", k=1)

5.1.6.10 删除文档（可选）

# 获取文档 ID（通常在 add_documents 时可指定或返回）
ids = vector_store.get()["ids"]   # 获取所有 ID
vector_store.delete(ids=ids[:1])  # 删除第一个文档

5.1.7 关键说明

方法/属性	作用
Chroma(embedding_function)	初始化向量库。persist_directory 指定磁盘存储路径，不设置则仅内存。
add_documents(docs)	添加 Document 对象列表（每个对象包含 page_content 和 metadata）。
add_texts(texts, metadatas)	直接添加文本列表，metadatas 为可选元数据列表，长度需与 texts 一致。
similarity_search(query,k)	返回与查询最相似的 k 个文档（无分数）。
similarity_search_with_score	返回 (Document, score) 列表，分数越小越相似（通常为欧氏距离或余弦距离）。
as_retriever(search_type)	将向量库包装为检索器（Runnable 接口），支持 similarity 或 mmr 检索类型。
search_kwargs	MMR 参数：k 最终返回数，fetch_k 初始检索数，lambda_mult 相关性 vs 多样性(0~1)。

5.2 FIASS

5.2.1 FAISS使用流程

FAISS 使用流程
1、创建：FAISS.from_texts(texts, embeddings)
2、保存：policy_db.save_local(“path”)
3、加载：FAISS.load_local(folder_path, embeddings, allow_dangerous_deserialization=True)
4、**更新/删除：**不支持直接修改，通常需重建整个索引。

多知识库管理

• 将不同来源、不同主题的文档分库存储（如每个文件一个独立索引）。

• 优点：更新独立、检索效率高、避免跨领域干扰。

5.2.3FAISS 主要痛点

痛点	说明
无法直接删除/修改	不支持原子化增删改，更新数据需要重建整个索引，大索引重建耗时数十分钟，影响服务可用性。
复杂逻辑需“手撸”	只返回向量 ID 和距离，不存储文档内容、来源等元数据。要实现“只搜最近7天文档”等逻辑，必须额外维护外部数据库（如 SQLite），增加系统复杂度。
重启重建耗时	基于内存运行，程序重启后必须重新加载数据并训练索引，等待时间长，且重启期间服务不可用。

对比：Chroma 支持持久化、元数据存储、动态增删改，更适合生产环境。

5.2.4代码示例（向量存储服务类）

class VectorStoreService:
    def __init__(self):
        self.embeddings = DashScopeEmbeddings(model='text-embedding-v1', api_key=config.API_KEY)
        self.index_path = None

    def update(self, texts, path):
        # 添加数据并保存
        self.FAISSdb.update_from_texts(texts)
        self.FAISSdb.save_local(path)

    def delete(self, path):
        # 删除数据（实际需重建索引）
        self.FAISSdb.delete_from_texts([])
        self.FAISSdb.save_local(path)

6. 数据分割

RecursiveCharacterTextSplitter 是处理通用文本时官方推荐的首选分割器。它采用递归方式尝试不同分隔符，力求在不破坏语义结构的前提下将文本分割成合适大小的块。

from langchain_text_splitters import RecursiveCharacterTextSplitter

# 配置分割器，设置块大小和重叠量
splitter = RecursiveCharacterTextSplitter(
    chunk_size=150,       # 目标块大小
    chunk_overlap=20,     # 重叠部分
    separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""]
)

# 待分割的长文本
long_text = """
这是第一段落。它包含了多个句子。我们希望能保持它的结构。

这是第二段落的开始。这段文字比较长，可能会触发更细粒度的分割。
"""

# 执行分割，返回字符串列表
chunks = splitter.split_text(long_text)

# 如果你处理的是 LangChain 的文档对象列表
# docs = splitter.split_documents(documents)

7.Memory

7.1Memory 临时会话

1. 短期记忆（Session-based Memory）

• **定义：**在一次会话（session）内，模型能记住用户之前说过的话，实现多轮对话的上下文连贯。
• **实现方式：**通过 RunnableWithMessageHistory 包装原始链，并为每个 session_id 维护独立的 BaseChatMessageHistory 对象（这里用了 InMemoryChatMessageHistory）。

2. RunnableWithMessageHistory

• 作用：给普通的 LCEL 链（Runnable）增加对话历史管理能力。
• 关键参数：
• runnable：原始链（如 prompt | model | parser）
• get_session_history：一个可调用对象，接收 session_id: str，返回 BaseChatMessageHistory 实例
• input_messages_key：输入字典中用户当前问题的键名（例如 “input”）
• history_messages_key：输入字典中历史消息占位符的键名（需与 MessagesPlaceholder 变量名一致）

3. 工作流程：

• 调用 conversation.invoke({“input”: “xxx”}, config={“configurable”: {“session_id”: “123”}})
• 根据 session_id 获取对应的历史存储对象
• 从存储对象中读取历史消息列表
• 将历史消息插入到提示词的 MessagesPlaceholder 位置
• 构造完整输入（历史 + 当前用户消息）调用原始链
• 将新产生的 AIMessage 追加到历史存储中

4.基础代码：

"""

短期记忆:
    当前对话的短期会话 一次会话中的多轮短期会话
"""
from fastapi import FastAPI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory
from langchain_deepseek import ChatDeepSeek
from config_data import config
from langchain_core.output_parsers import StrOutputParser
app = FastAPI()
class  SessionHistoryManager:
    def __init__(self):
        self.store = {}
    def get_session_history(self, session_id: str) -> BaseChatMessageHistory:
        if session_id not in self.store:
            self.store[session_id] = InMemoryChatMessageHistory()
        return self.store[session_id]

history_manager = SessionHistoryManager()
# 提示词模板
prompts=ChatPromptTemplate.from_messages([
   ("system", "你需要根据会话历史回答用户问题。"),
    MessagesPlaceholder(variable_name="chat_history"),  # 关键：占位符用于插入历史消息
    ("user", "请回答用户提问：{input}")
])
models=ChatDeepSeek(
    model_name=config.DEEPSEEK_MODEL,
    api_key=config.DEEPSEEK_API_KEY,
    api_base=config.DEEPSEEK_API_URL,
    temperature=0.5,
    max_tokens=1024,
    top_p=1
)
base_chain= prompts| models | StrOutputParser()
conversation=RunnableWithMessageHistory(
    base_chain,
    history_manager.get_session_history,  # 传入方法，不是实例
    input_messages_key="input",       
    history_messages_key="chat_history" 
)

@app.post("/chat")
async def chat(input: str, session_id: str):
    try:
        result = await conversation.ainvoke(   # 使用 ainvoke
            {"input": input},
            config={"configurable": {"session_id": session_id}}
        )
        return {"response": result}
    except Exception as e:
        return {"error": str(e)}

if __name__ == '__main__':
    print("starting server...")
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)

7.2 长期会话

8.LangchainV1.下与V0.3 比较

8.1统一Agent创建接口 create_agent()

LLM智能体循环运行各种工具实现目标。智能体持续运行，知道满足条件为止（即模型发出最终输出或达到迭代次数限制）
create_anget提供生成环境的代理实现，使用langGraph构建基于图的代理运行时

from langchain.agents import create_agent
create_agent(
"模型"，
tools=tools
)

8.2 中间件系统（Middleware）

什么是中间件
是一种流程控制机制，用于智能体执行过程中拦截，修改或增强请求与响应的处理逻辑，而无需修改核心Agent或工具的代码
能做什么
监视器、控制、执行、调整等
中间件举例
总结摘要中间件（上下文压缩）:当竭尽会话次数上限时，自动汇总对话历史记录

agent=create_agent(
model='',
tools=tools,
SummarizationMiddleware(
	model='',
)
)

1.内置中间件
2.自定义中间件
自定义中间件：通过在代理执行流程的特定点运行狗子来构建自定义中间件
基于装饰器的:适用于单钩中间件的快速简便
基于类的：对于具有多个钩子的复杂中间件来说功能更强大

8.2.3 自定义中间件

8.2.3.1 围绕模型-修饰器

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain.agents.middleware import wrap_model_call,ModelRequest,ModelResponse
from langgraph.checkpoint.memory import InMemorySaver  
from typing import Callable
base_model = ChatOpenAI(temperature=0, model="gpt-3.5-turbo")  # 基础模型
model = ChatOpenAI(temperature=0, model="gpt-4")  # 高级模型

#围绕每次模型调用 如果请求大于5次 切换模型
@wrap_model_call
def switch_model(request:ModelRequest,handler:Callable[[ModelRequest],ModelResponse])->ModelResponse:
    model=base_model
    """根据对话轮次动态选择模型"""
    message_count = len(request.state["messages"])
    if message_count > 10:
        print("检测到长对话切换模型")
        return handler(request.override(model=model))
    else:
        return handler(request)

agents=create_agent(
    model=base_model,
    tools=[],
    middleware=[switch_model],
    checkpointer=InMemorySaver(),
)
config={"configurable":{"thread_id":"1111"}}
chunks=agents.stream({"messages":[{'role':'user','content':'hello word'}]},config=config)
for chunk in chunks:
    print(chunk)

8.2.3.2 围绕模型-类

from langchain.agents import create_agent
from langchain_deepseek import ChatDeepSeek
from langgraph.checkpoint.memory import InMemorySaver  
from langchain.agents.middleware import AgentMiddleware ,ModelRequest,ModelResponse
from langchain_openai import ChatOpenAI
from typing import Callable

base_model = ChatOpenAI(temperature=0, model="gpt-3.5-turbo")  # 基础模型
model = ChatDeepSeek(temperature=0,  model="deepseek-chat")  # 高级模型

class SwitchModel(AgentMiddleware):
    def __init__(self, model: Callable, base_model: Callable):
        super().__init__()
        self.model = model
        self.base_model = base_model
    def wrap_model_call(
        self, 
        request: ModelRequest, 
        handler: Callable[[ModelRequest], ModelResponse]
    ) -> ModelResponse:
        message_count = len(request.state["messages"])  # 注意是 request.state["messages"]
        if message_count > 10:
            print(f"切换到高级模型（消息数: {message_count}）")
            return handler(request.override(model=self.model))
        else:
            print(f"使用基础模型（消息数: {message_count}）")
            return handler(request)


switchModel = SwitchModel(model, base_model)
agent =create_agent(
    model=base_model,
    tools=[],
    middleware=[switchModel],
    checkpointer=InMemorySaver(),
)
# 添加短期记忆功能
config={"configurable":{"thread_id":"1111"}}
chunks=agent.stream(
    {"messages":[{'role':'user','content':'hello word'}]},
    config=config
)

3.标准输出
4.结构化输出

8.3 Langchain是一个构建LLM应用的核心框架

9.Langchin 实现RAG(检索增强生成）

1.索引化
将整个文档进行切块（完整/语义）-——>将切块成文档嵌入模型---->进行索引化–》文档+向量+id+元数据
2.检索、查询

query—>Encode—问题向量化–>去向量数据库查询->返回结果–>加入提示词|大模型进行查询——>返回结果

#10 召回
召回：其实就是去向量数据库进行检索，
检索的过程：召回->精排->生成

10.1 多路召回概述

多路召回概述
多路召回是结合多种召回策略的结果，如向量召回基于语义相似度，关键词召回基于传统匹配算法，融合多路结果得到更全面准确的候选文档集合，
重要性：
在RAG架构中，多路召回能够提高信息检索的效率和准确性，为后续的生产环境提供更丰富的上下文信息，从而生成更高质量的回答
应用场景
广泛引用与搜索引擎、推荐系统】智能客服等领域、尤其在医疗、金融等对信息准确性要求较高的领域，多路召回能够更好地满足用户需求

多路召回方式
并行多路召回：将不同召回策略的结果分别获取进行合并去重，在通过重排序综合打分 ，综合打分，去最终的Top N结果。例如向量召回Top 10，，BM25文本召回Top 10，合并后重排序取Top10
优势：充分利用语义和关键词两种优势，互补性强，能够更全面的覆盖候选文档，减少因单一招呼策略导致落检问题，是当前工业界最流行的做法
串行多路召回
实现方式：
串行召回是先用一路召回大量粗筛结果，在用另一路算发粗筛结果做精排。
例如先用向量召回100条，再用BM25重新打分排序取Top 10. 页可以反过来操作
适用场景
当数据量较大且计算资源有限时，串行多路召回可以有效减少计算量，先通过粗筛快速缩小范围，再通过精排提高结果质量。
优缺点
优点是计算效率较高，缺点是如果粗筛阶段筛选不准确，可能会导致后续精排阶段丢失重要信息，影响最终结果的准确性。