LangChain Chat Model学习笔记-EW帮帮网

Prompt templates: Few shot、Example selector

一、Few shot(少量示例)

创建少量示例的格式化程序

创建一个简单的提示模板，用于在生成时向模型提供示例输入和输出。向LLM提供少量这样的示例被称为少量示例，这是一种简单但强大的指导生成的方式，在某些情况下可以显著提高模型性能。
少量示例提示模板可以由一组示例或一个负责从定义的集合中选择一部分示例的示例选择器类构建。
配置一个格式化程序，将少量示例格式化为字符串。这个格式化程序应该是一个PromptTemplate对象。

from langchain_core.prompts import PromptTemplate
example_prompt = PromptTemplate.from_template("问题：{question}\n{answer}")

创建示例集合
接下来，我们将创建一个少量示例的列表。每个示例应该是一个字典，表示我们上面定义的格式化提示的示例输入。

examples = [
    {
        "question": "谁活得更长，穆罕默德·阿里还是艾伦·图灵？",
        "answer": """
是否需要后续问题：是的。
后续问题：穆罕默德·阿里去世时多大年纪？
中间答案：穆罕默德·阿里去世时74岁。
后续问题：艾伦·图灵去世时多大年纪？
中间答案：艾伦·图灵去世时41岁。
所以最终答案是：穆罕默德·阿里
""",
    },
    {
        "question": "克雷格斯列表的创始人是什么时候出生的？",
        "answer": """
是否需要后续问题：是的。
后续问题：克雷格斯列表的创始人是谁？
中间答案：克雷格斯列表的创始人是克雷格·纽马克。
后续问题：克雷格·纽马克是什么时候出生的？
中间答案：克雷格·纽马克于1952年12月6日出生。
所以最终答案是：1952年12月6日
""",
    },
    {
        "question": "乔治·华盛顿的外祖父是谁？",
        "answer": """
是否需要后续问题：是的。
后续问题：乔治·华盛顿的母亲是谁？
中间答案：乔治·华盛顿的母亲是玛丽·波尔·华盛顿。
后续问题：玛丽·波尔·华盛顿的父亲是谁？
中间答案：玛丽·波尔·华盛顿的父亲是约瑟夫·波尔。
所以最终答案是：约瑟夫·波尔
""",
    },
    {
        "question": "《大白鲨》和《皇家赌场》的导演都来自同一个国家吗？",
        "answer": """
是否需要后续问题：是的。
后续问题：《大白鲨》的导演是谁？
中间答案：《大白鲨》的导演是史蒂文·斯皮尔伯格。
后续问题：史蒂文·斯皮尔伯格来自哪个国家？
中间答案：美国。
后续问题：《皇家赌场》的导演是谁？
中间答案：《皇家赌场》的导演是马丁·坎贝尔。
后续问题：马丁·坎贝尔来自哪个国家？
中间答案：新西兰。
所以最终答案是：不是
""",
    },
]

让我们使用其中一个示例测试格式化提示：

print(example_prompt.invoke(examples[0]).to_string())

问题：谁活得更长，穆罕默德·阿里还是艾伦·图灵？

是否需要后续问题：是的。
后续问题：穆罕默德·阿里去世时多大年纪？
中间答案：穆罕默德·阿里去世时74岁。
后续问题：艾伦·图灵去世时多大年纪？
中间答案：艾伦·图灵去世时41岁。
所以最终答案是：穆罕默德·阿里

将示例和格式化程序传递给FewShotPromptTemplate

最后，创建一个FewShotPromptTemplate对象。该对象接受少量示例和少量示例的格式化程序。当格式化此FewShotPromptTemplate时，它使用example_prompt格式化传递的示例，然后将它们添加到suffix之前的最终提示中：

from langchain_core.prompts import FewShotPromptTemplate
prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_prompt,
    suffix="问题：{input}",
    input_variables=["input"],
)
print(
    prompt.invoke({"input": "乔治·华盛顿的父亲是谁？"}).to_string()
)

问题：谁活得更长，穆罕默德·阿里还是艾伦·图灵？

是否需要后续问题：是的。
后续问题：穆罕默德·阿里去世时多大年纪？
中间答案：穆罕默德·阿里去世时74岁。
后续问题：艾伦·图灵去世时多大年纪？
中间答案：艾伦·图灵去世时41岁。
所以最终答案是：穆罕默德·阿里


问题：克雷格斯列表的创始人是什么时候出生的？

是否需要后续问题：是的。
后续问题：克雷格斯列表的创始人是谁？
中间答案：克雷格斯列表的创始人是克雷格·纽马克。
后续问题：克雷格·纽马克是什么时候出生的？
中间答案：克雷格·纽马克于1952年12月6日出生。
所以最终答案是：1952年12月6日


问题：乔治·华盛顿的外祖父是谁？

是否需要后续问题：是的。
后续问题：乔治·华盛顿的母亲是谁？
中间答案：乔治·华盛顿的母亲是玛丽·波尔·华盛顿。
后续问题：玛丽·波尔·华盛顿的父亲是谁？
中间答案：玛丽·波尔·华盛顿的父亲是约瑟夫·波尔。
所以最终答案是：约瑟夫·波尔


问题：《大白鲨》和《皇家赌场》的导演都来自同一个国家吗？

是否需要后续问题：是的。
后续问题：《大白鲨》的导演是谁？
中间答案：《大白鲨》的导演是史蒂文·斯皮尔伯格。
后续问题：史蒂文·斯皮尔伯格来自哪个国家？
中间答案：美国。
后续问题：《皇家赌场》的导演是谁？
中间答案：《皇家赌场》的导演是马丁·坎贝尔。
后续问题：马丁·坎贝尔来自哪个国家？
中间答案：新西兰。
所以最终答案是：不是


问题：乔治·华盛顿的父亲是谁？

通过向模型提供这样的示例，我们可以引导模型做出更好的回应。

Example selectors(示例选择器)

我们将重用上一节中的示例集和格式化程序。但是，我们不会直接将示例馈送到 FewShotPromptTemplate 对象中，而是将它们馈送到名为 SemanticSimilarityExampleSelector 的 ExampleSelector 实现实例中。该类根据输入与少样本示例的相似性选择初始集合中的少样本示例。它使用嵌入模型计算输入与少样本示例之间的相似性，以及向量存储库执行最近邻搜索。
为了展示它的样子，让我们初始化一个实例并在隔离环境中调用它：

from langchain_chroma import Chroma
from langchain_core.example_selectors import SemanticSimilarityExampleSelector
from langchain_openai import OpenAIEmbeddings
example_selector = SemanticSimilarityExampleSelector.from_examples(
    # 这是可供选择的示例列表。
    examples,
    # 这是用于生成嵌入的嵌入类，用于衡量语义相似性。
    OpenAIEmbeddings(),
    # 这是用于存储嵌入并进行相似性搜索的 VectorStore 类。
    Chroma,
    # 这是要生成的示例数量。
    k=1,
)
# 选择与输入最相似的示例。
question = "玛丽·波尔·华盛顿的父亲是谁？"
selected_examples = example_selector.select_examples({"question": question})
print(f"与输入最相似的示例: {question}")
for example in selected_examples:
    print("\n")
    for k, v in example.items():
        print(f"{k}: {v}")

Chroma 安装报错

pip install langchain-chroma
error: Microsoft Visual C++ 14.0 or greater is required. Get it with "Microsoft C++ Build Tools": https://visualstudio.microsoft.com/visual-cpp-build-tools/
[end of output]

解决方案：需要先下载visual-cpp-build-tools，再执行pip install langchain-chroma
下载地址：https://visualstudio.microsoft.com/zh-hans/visual-cpp-build-tools/
在这里插入图片描述
现在，让我们创建一个 FewShotPromptTemplate 对象。该对象接受示例选择器和用于少样本示例的格式化程序提示。

prompt = FewShotPromptTemplate(
    example_selector=example_selector,
    example_prompt=example_prompt,
    suffix="Question: {input}",
    input_variables=["input"],
)
print(
    prompt.invoke({"input": "玛丽·波尔·华盛顿的父亲是谁？"}).to_string()
)

问题：乔治·华盛顿的外祖父是谁？
是否需要后续问题：是的。
后续问题：乔治·华盛顿的母亲是谁？
中间答案：乔治·华盛顿的母亲是玛丽·波尔·华盛顿。
后续问题：玛丽·波尔·华盛顿的父亲是谁？
中间答案：玛丽·波尔·华盛顿的父亲是约瑟夫·波尔。
所以最终答案是：约瑟夫·波尔
Question: 玛丽·波尔·华盛顿的父亲是谁？

二、LangServe

概述

LangServe 🦜️🏓 帮助开发者将 LangChain 可运行和链部署为 REST API。
该库集成了 FastAPI 并使用 pydantic 进行数据验证。
Pydantic 是一个在 Python中用于数据验证和解析的第三方库，现在是Python中使用广泛的数据验证库。
● 它利用声明式的方式定义数据模型和Python 类型提示的强大功能来执行数据验证和序列化，使您的代码更可靠、更可读、更简洁且更易于调试。。
● 它还可以从模型生成 JSON 架构，提供了自动生成文档等功能，从而轻松与其他工具集成
此外，它提供了一个客户端，可用于调用部署在服务器上的可运行对象。JavaScript 客户端可在 LangChain.js 中找到。
特性
● 从 LangChain 对象自动推断输入和输出模式，并在每次 API 调用中执行，提供丰富的错误信息
● 带有 JSONSchema 和 Swagger 的 API 文档页面（插入示例链接）
● 高效的 /invoke、/batch 和 /stream 端点，支持单个服务器上的多个并发请求
● /stream_log 端点，用于流式传输链/代理的所有（或部分）中间步骤
● 新功能自 0.0.40 版本起，支持 /stream_events，使流式传输更加简便，无需解析 /stream_log 的输出。
● 使用经过严格测试的开源 Python 库构建，如 FastAPI、Pydantic、uvloop 和 asyncio。
● 使用客户端 SDK 调用 LangServe 服务器，就像本地运行可运行对象一样（或直接调用 HTTP API）

三、什么是LangServe ？

LangServe 是一个利用 FastAPI 和 Swagger 开发的服务，旨在帮助开发者将 LangChain 的功能部署为可运行的 REST API。LangChain 是一个强大的工具，用于构建和部署复杂的语言模型链，这些链可以执行一系列任务，从数据提取到生成文本。通过将 LangChain 部署为 REST API，开发者可以轻松地集成这些功能到他们的应用程序中，无论是网站、移动应用还是其他服务。

以下是 LangServe 的一些关键特性和优势：

基于 FastAPI：FastAPI 是一个现代的、快速（高性能）的 Web 框架，用于构建 API。它基于 Python 类型提示，自动生成 OpenAPI 和 JSON Schema 文档，这意味着你可以轻松地为你的 API 创建交互式文档，而无需额外的工作。
集成 Swagger：Swagger（或 OpenAPI）提供了 API 的标准规范，允许开发者设计、构建、记录和使用 RESTful Web 服务。LangServe 通过集成 Swagger，提供了一个用户友好的界面来探索和测试 API 端点。
易于部署：LangServe 使得将 LangChain 链部署为 REST API 变得简单直接。开发者无需深入了解复杂的后端设置，即可快速启动和运行他们的语言模型链。
可扩展性：由于 FastAPI 的高性能和模块化设计，LangServe 能够轻松扩展以满足不断增长的需求。无论是增加新的 API 端点还是优化现有功能，LangServe 都提供了足够的灵活性。
社区支持：作为开源项目，LangServe 和其依赖的 FastAPI、Swagger 等技术都有庞大的社区支持。这意味着开发者在遇到问题时，可以很容易地找到解决方案或寻求帮助。
安全性：虽然具体的安全实现取决于部署配置，但 FastAPI 提供了多种内置的安全特性，如认证和授权机制。开发者可以利用这些特性来确保他们的 LangServe 实例的安全性。
总之，LangServe 是一个强大的工具，它结合了 FastAPI 的高性能和 Swagger 的易用性，为开发者提供了一个简单而有效的方式来将 LangChain 的功能部署为 REST API。这使得集成和使用复杂的语言模型链变得更加容易和高效。

FastAPI 是一个用于构建 API 的现代、快速（高性能）的 Web 框架，它是基于 Python 3.6+ 的类型提示构建的。FastAPI 的核心特性之一是其能够自动生成符合 OpenAPI 规范（也称作 Swagger 或 OpenAPI 3.0）和 JSON Schema 的 API 文档。这意味着，当你使用 FastAPI 构建 API 时，你可以立即获得一个交互式的 API 文档界面，无需编写额外的文档代码。

四、FastAPI是什么

FastAPI 的主要优点包括：

快速开发：由于其基于 Python 类型提示的设计，FastAPI 允许开发者以极少的代码快速定义 API 路由、请求和响应模型。
高性能：FastAPI 使用 Starlette 和 Pydantic 作为其核心依赖项，这些库都是异步的，因此 FastAPI 能够充分利用现代 Python 异步编程的特性，提供高性能的 API 服务。
自动生成文档：如前面所述，FastAPI 能够自动生成 OpenAPI 和 JSON Schema 文档，这些文档对于 API 的用户来说是非常有用的，因为它们提供了 API 的完整描述，包括端点、请求参数、响应格式等。
易于测试和调试：由于 FastAPI 提供了清晰的 API 定义和自动生成的文档，开发者可以更容易地测试和调试他们的 API。
可扩展性：FastAPI 的模块化设计使其易于扩展。开发者可以轻松地添加中间件、自定义路由处理器、错误处理器等，以满足他们的特定需求。
社区支持：作为一个流行的开源项目，FastAPI 拥有一个活跃的社区，这意味着开发者在遇到问题时可以获得及时的帮助和支持。
与现有工具和库的兼容性：FastAPI 与许多现有的 Python 库和工具兼容，这使得它很容易集成到现有的项目中。

总的来说，FastAPI 是一个强大而灵活的 Web 框架，它提供了快速开发、高性能和自动生成文档等关键特性，使得构建和维护 API 变得更加容易和高效。

FastAPI 和Java的什么框架类似？

FastAPI作为一个基于Python的Web框架，与Java中的一些框架在功能和设计理念上有一定的相似性，但具体到某一个Java框架进行类比并非完全准确，因为每个框架都有其独特的特性和应用场景。不过，如果要从整体架构、开发效率、性能以及自动生成文档等方面来看，FastAPI可以与Java中的以下框架进行类比：

Spring Boot：
- 开发效率：Spring Boot提供了大量的自动配置和约定优于配置的理念，使得开发者能够快速搭建和启动Web应用。同样，FastAPI也注重开发效率，通过基于Python类型提示的自动文档生成和路由定义，简化了API的开发过程。
- 性能：Spring Boot支持异步编程，并且可以通过各种优化手段提高性能。FastAPI则基于Starlette框架构建，天生支持异步编程，能够高效处理并发请求。
- 自动生成文档：Spring Boot可以通过Spring Fox或Springdoc等工具集成Swagger，自动生成API文档。FastAPI则内置了OpenAPI和JSON Schema的自动生成功能，提供了交互式的API文档界面。
JAX-RS（Java API for RESTful Web Services）：
- RESTful API开发：JAX-RS是Java中用于开发RESTful Web服务的标准API。FastAPI同样专注于RESTful API的开发，提供了简洁的路由定义和请求/响应处理机制。
- 异步支持：虽然JAX-RS本身并不直接支持异步编程，但Java EE中的其他组件（如Servlet 3.0+的异步支持）可以与之结合使用实现异步处理。而FastAPI则天生支持异步编程，使得处理I/O密集型任务更加高效。

需要注意的是，尽管FastAPI与上述Java框架在某些方面有相似性，但它们各自的语言生态、社区支持、以及与其他技术的集成方式等方面都存在差异。因此，在选择使用哪个框架时，开发者需要根据自己的项目需求、技术栈以及个人偏好进行综合考虑。

总的来说，FastAPI作为一个现代化的Python Web框架，在开发效率、性能和自动生成文档等方面表现出色，与Java中的Spring Boot和JAX-RS等框架在功能和设计理念上有一定的相似性。

五、限制

● 目前不支持服务器发起的事件的客户端回调
● 当使用 Pydantic V2 时，将不会生成 OpenAPI 文档。FastAPI 支持混合使用 pydantic v1 和 v2 命名空间。更多细节请参见下面的章节。

六、如何创建一个LangServe应用？

更新到最新版本

安装

对于客户端和服务器：

pip install --upgrade "langserve[all]" # 如下代表成功

在这里插入图片描述
或者对于客户端代码，pip install "langserve[client]"，对于服务器代码，pip install "langserve[server]"。

七、LangChain CLI 🛠️

使用 LangChain CLI 快速启动 LangServe 项目。
要使用 langchain CLI，请确保已安装最新版本的 langchain-cli。您可以使用 pip install -U langchain-cli 进行安装。
设置
注意：我们使用 poetry 进行依赖管理。请参阅 poetry 文档了解更多信息。

使用 langchain cli 命令创建新应用

langchain app new my-app   # 一直回车就行

在这里插入图片描述

2.可在生成的代码add_routes 中定义可运行对象。找到 server.py 并编辑

add_routes(app. NotImplemented)
例如：
add_routes(
    app,
    ChatOpenAI(model="gpt-3.5-turbo"),
    path="/openai",
)

使用 poetry 添加第三方包（例如 langchain-openai、langchain-anthropic、langchain-mistral 等）

#安装pipx，参考：https://pipx.pypa.io/stable/installation/
pip install pipx 
#加入到环境变量，需要重启PyCharm 
pipx ensurepath

# 安装poetry，参考：https://python-poetry.org/docs/
pipx install poetry


#安装 langchain-openai 库，例如：poetry add [package-name]
poetry add langchain
poetry add langchain-openai

其实也可以直接在pyproject.toml文件指定对应的版本就行


[tool.poetry.dependencies]
python = "^3.11"
uvicorn = "^0.23.2"
langserve = {extras = ["server"], version = ">=0.0.30"}
pydantic = ">=2.7.4"
langchain = "^0.3.4"
langchain-openai = "^0.2.3"
它类似于java的pom.xml文件

设置相关环境变量。例如，

export OPENAI_API_KEY="sk-..."

启动您的应用

poetry run langchain serve --port=8000

在这里插入图片描述
如果您是这样关闭的：

同一个端口反复启动，容易被占用。会导致页面访问空白。要么ctrl+x停止，要么换端口。

八、示例应用

服务器

以下是一个部署 OpenAI 聊天模型，讲述有关特定主题笑话的链的服务器。

#!/usr/bin/env python
from fastapi import FastAPI
from langchain_openai import ChatOpenAI
from langserve import add_routes
app = FastAPI(
    title="LangChain 服务器",
    version="1.0",
    description="使用 Langchain 的 Runnable 接口的简单 API 服务器",
)
add_routes(
    app,
    ChatOpenAI(model="gpt-4"),
    path="/openai",
)
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="localhost", port=8000)

如果您打算从浏览器调用您的端点，您还需要设置 CORS 头。
您可以使用 FastAPI 的内置中间件来实现：

from fastapi.middleware.cors import CORSMiddleware
# 设置所有启用 CORS 的来源
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["*"],
)

文档

如果您已部署上述服务器，可以使用以下命令查看生成的 OpenAPI 文档：
文档地址：http://localhost:8000/docs

curl localhost:8000/docs

请确保添加 /docs 后缀。
⚠️ 首页 / 没有被设计定义，因此 curl localhost:8000 或访问该 URL
将返回 404。如果您想在 / 上有内容，请定义一个端点 @app.get(“/”)。

九、客户端

Python SDK

from langchain.schema.runnable import RunnableMap
from langchain_core.prompts import ChatPromptTemplate
from langserve import RemoteRunnable

openai = RemoteRunnable("http://localhost:8000/openai/")
prompt = ChatPromptTemplate.from_messages(
    [("system", "你是一个喜欢写故事的助手"), ("system", "写一个故事，主题是： {topic}")]
)
# 可以定义自定义链
chain = prompt | RunnableMap({
    "openai": openai
})
response = chain.batch([{"topic": "猫"}])
print(response)
#[{'openai': AIMessage(content='从前，有一个叫做肖恩的男孩，他在一个宁静的乡村里生活。一天，他在家的后院发现了一个小小的，萌萌的猫咪。这只猫咪有一双大大的蓝色眼睛，毛色如同朝霞般的粉色，看起来非常可爱。\n\n肖恩把这只猫咪带回了家，他给她取名为“樱花”，因为她的毛色让他联想到春天盛开的樱花。肖恩非常喜欢樱花，他用心照顾她，每天都会为她准备新鲜的食物和清水，还会陪她玩耍，带她去散步。\n\n樱花也非常喜欢肖恩，她会在肖恩读书的时候躺在他的脚边，会在他伤心的时候安慰他，每当肖恩回家的时候，她总是第一个跑出来迎接他。可是，樱花有一个秘密，她其实是一只会说人话的猫。\n\n这个秘密是在一个月圆的夜晚被肖恩发现的。那天晚上，肖恩做了一个噩梦，他从梦中惊醒，发现樱花正坐在他的床边，用人的语言安慰他。肖恩一开始以为自己在做梦，但是当他清醒过来，樱花还在继续讲话，他才知道这是真的。\n\n樱花向肖恩解释，她是一只来自神秘的猫咪国度的使者，她的任务是保护和帮助那些善良和爱护动物的人。肖恩因为对她的善良和照顾，使她决定向他展现真实的自我。\n\n肖恩虽然感到惊讶，但他并没有因此而害怕或者排斥樱花。他觉得这只使得他更加喜欢樱花，觉得这是他们之间的特殊纽带。\n\n从那天开始，肖恩和樱花的关系变得更加亲密，他们像最好的朋友一样，分享彼此的秘密，一起度过快乐的时光。樱花也用她的智慧和力量，帮助肖恩解决了许多困扰他的问题。\n\n许多年过去了，肖恩长大了，他离开了乡村，去了城市上大学。但是，无论他走到哪里，都会带着樱花。他们的友情和互相的陪伴，让他们无论在哪里，都能感到家的温暖。\n\n最后，肖恩成为了一名作家，他写下了自己和樱花的故事，这个故事被人们广为传播，让更多的人知道了这个关于善良、友情和勇气的故事。而樱花，也永远陪伴在肖恩的身边，成为他生活中不可或缺的一部分。\n\n这就是肖恩和樱花的故事，一个关于男孩和他的猫的故事，充满了奇迹、温暖和爱。', response_metadata={'token_usage': {'completion_tokens': 1050, 'prompt_tokens': 33, 'total_tokens': 1083}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None}, id='run-c44f1624-ea75-424b-ba3d-e741baf44bda-0', usage_metadata={'input_tokens': 33, 'output_tokens': 1050, 'total_tokens': 1083})}]

在 TypeScript 中（需要 LangChain.js 版本 0.0.166 或更高）：

import { RemoteRunnable } from "@langchain/core/runnables/remote";
const chain = new RemoteRunnable({
  url: `http://localhost:8000/openai/`,
});
const result = await chain.invoke({
  topic: "cats",
});

使用 requests 的 Python 代码：

import requests
response = requests.post(
    "http://localhost:8000/openai",
    json={'input': {'topic': 'cats'}}
)
response.json()

您也可以使用 curl：

curl --location --request POST 'http://localhost:8000/openai/stream' \
    --header 'Content-Type: application/json' \
    --data-raw '{
        "input": {
            "topic": "狗"
        }
    }'

十、端点（也就是路由）

以下代码：

add_routes(
    app,
    runnable,
    path="/my_runnable",
)

将以下端点添加到服务器：
● POST /my_runnable/invoke - 对单个输入调用可运行项
● POST /my_runnable/batch - 对一批输入调用可运行项
● POST /my_runnable/stream - 对单个输入调用并流式传输输出
● POST /my_runnable/stream_log - 对单个输入调用并流式传输输出，
包括生成的中间步骤的输出
● POST /my_runnable/astream_events - 对单个输入调用并在生成时流式传输事件，
包括来自中间步骤的事件。
● GET /my_runnable/input_schema - 可运行项的输入的 JSON 模式
● GET /my_runnable/output_schema - 可运行项的输出的 JSON 模式
● GET /my_runnable/config_schema - 可运行项的配置的 JSON 模式
这些端点与LangChain 表达式语言接口相匹配 –

十一、为 Chain 添加 Message history (Memory)单行初始化 chat model

对话状态Chain传递

在构建聊天机器人时，将对话状态传递到链中以及从链中传出对话状态至关重要。RunnableWithMessageHistory 类让我们能够向某些类型的链中添加消息历史。它包装另一个 Runnable 并管理其聊天消息历史。
具体来说，它可用于任何接受以下之一作为输入的 Runnable：
● 一系列 BaseMessages
● 具有以序列 BaseMessages 作为值的键的字典
● 具有以字符串或序列 BaseMessages 作为最新消息的值的键和一个接受历史消息的单独键的字典
并将以下之一作为输出返回：
● 可视为 AIMessage 内容的字符串
● 一系列 BaseMessage
● 具有包含一系列 BaseMessage 的键的字典
让我们看一些示例以了解其工作原理。首先，我们构建一个 Runnable（此处接受字典作为输入并返回消息作为输出）：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai.chat_models import ChatOpenAI
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You're an assistant who's good at {ability}. Respond in 20 words or fewer",
        ),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),
    ]
)
runnable = prompt | model

first=ChatPromptTemplate(input_variables=['ability', 'history', 'input'], input_types={'history': typing.List[typing.Union[langchain_core.messages.ai.AIMessage, langchain_core.messages.human.HumanMessage, langchain_core.messages.chat.ChatMessage, langchain_core.messages.system.SystemMessage, langchain_core.messages.function.FunctionMessage, langchain_core.messages.tool.ToolMessage]]}, messages=[SystemMessagePromptTemplate(prompt=PromptTemplate(input_variables=['ability'], template="You're an assistant who's good at {ability}. Respond in 20 words or fewer")), MessagesPlaceholder(variable_name='history'), HumanMessagePromptTemplate(prompt=PromptTemplate(input_variables=['input'], template='{input}'))]) last=ChatOpenAI(client=<openai.resources.chat.completions.Completions object at 0x0000026A478DB440>, async_client=<openai.resources.chat.completions.AsyncCompletions object at 0x0000026A478FCD10>, model_name='gpt-4', openai_api_key=SecretStr('**********'), openai_proxy='')

要管理消息历史，我们需要：

此 Runnable；
一个返回 BaseChatMessageHistory 实例的可调用对象。

十二、聊天历史存储在内存

下面我们展示一个简单的示例，其中聊天历史保存在内存中，此处通过全局 Python 字典实现。
我们构建一个名为 get_session_history 的可调用对象，引用此字典以返回 ChatMessageHistory 实例。通过在运行时向 RunnableWithMessageHistory 传递配置，可以指定可调用对象的参数。默认情况下，期望配置参数是一个字符串 session_id。可以通过 history_factory_config 关键字参数进行调整。
使用单参数默认值：

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
store = {}
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]
with_message_history = RunnableWithMessageHistory(
    runnable,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

请注意，我们已指定了 input_messages_key（要视为最新输入消息的键）和 history_messages_key（要添加历史消息的键）。
在调用此新 Runnable 时，我们通过配置参数指定相应的聊天历史：

with_message_history.invoke(
    {"ability": "math", "input": "余弦是什么意思？"},
    config={"configurable": {"session_id": "abc123"}},
)

# 记住
with_message_history.invoke(
    {"ability": "math", "input": "什么?"},
    config={"configurable": {"session_id": "abc123"}},
)

# 新的 session_id --> 不记得了。
with_message_history.invoke(
    {"ability": "math", "input": "什么?"},
    config={"configurable": {"session_id": "def234"}},
)

content='对不起，我没明白你的问题。你能再详细一点吗？我很擅长数学。' response_metadata={'token_usage': {'completion_tokens': 34, 'prompt_tokens': 32, 'total_tokens': 66}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-3f69d281-a850-452f-8055-df70d4936630-0' usage_metadata={'input_tokens': 32, 'output_tokens': 34, 'total_tokens': 66}

十三、基于 LangChain 的 Chatbot: Chat History

配置会话唯一键

我们可以通过向 history_factory_config 参数传递一个 ConfigurableFieldSpec 对象列表来自定义跟踪消息历史的配置参数。下面我们使用了两个参数：user_id 和 conversation_id。
配置user_id和conversation_id作为会话唯一键

from langchain_core.runnables import ConfigurableFieldSpec
store = {}
def get_session_history(user_id: str, conversation_id: str) -> BaseChatMessageHistory:
    if (user_id, conversation_id) not in store:
        store[(user_id, conversation_id)] = ChatMessageHistory()
    return store[(user_id, conversation_id)]
with_message_history = RunnableWithMessageHistory(
    runnable,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
    history_factory_config=[
        ConfigurableFieldSpec(
            id="user_id",
            annotation=str,
            name="User ID",
            description="用户的唯一标识符。",
            default="",
            is_shared=True,
        ),
        ConfigurableFieldSpec(
            id="conversation_id",
            annotation=str,
            name="Conversation ID",
            description="对话的唯一标识符。",
            default="",
            is_shared=True,
        ),
    ],
)
with_message_history.invoke(
    {"ability": "math", "input": "余弦是什么意思？"},
    config={"configurable": {"user_id": "123", "conversation_id": "1"}},
)

content='对不起，你能提供一些更详细的信息吗？我会很高兴帮助你解决数学问题。' response_metadata={'token_usage': {'completion_tokens': 38, 'prompt_tokens': 32, 'total_tokens': 70}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-02030348-7bbb-4f76-8c68-61785d012c26-0' usage_metadata={'input_tokens': 32, 'output_tokens': 38, 'total_tokens': 70}

在许多情况下，持久化对话历史是可取的。RunnableWithMessageHistory 对于 get_session_history 可调用如何检索其聊天消息历史是中立的。请参见这里，这是一个使用本地文件系统的示例。下面我们演示如何使用 Redis。请查看内存集成页面，以获取使用其他提供程序的聊天消息历史的实现。

十四、消息持久化

请查看 memory integrations 页面，了解使用 Redis 和其他提供程序实现聊天消息历史的方法。这里我们演示使用内存中的 ChatMessageHistory 以及使用 RedisChatMessageHistory 进行更持久存储。

配置redis环境

如果尚未安装 Redis，我们需要安装它：

pip install redis

%pip install --upgrade --quiet redis

如果我们没有现有的 Redis 部署可以连接，可以启动本地 Redis Stack 服务器：

docker run -d -p 6379:6379 -p 8001:8001 redis/redis-stack:latest

REDIS_URL = "redis://localhost:6379/0"

下载redis,解压运行redis-server
下载可视化工具RedisDesktopManager，连接redis,就可以看到相应得缓存信息

调用聊天接口，看Redis是否存储历史记录

更新消息历史实现只需要我们定义一个新的可调用对象，这次返回一个 RedisChatMessageHistory 实例：

from langchain_community.chat_message_histories import RedisChatMessageHistory
def get_message_history(session_id: str) -> RedisChatMessageHistory:
    return RedisChatMessageHistory(session_id, url=REDIS_URL)
with_message_history = RunnableWithMessageHistory(
    runnable,
    get_message_history,
    input_messages_key="input",
    history_messages_key="history",
)

我们可以像以前一样调用：

with_message_history.invoke(
    {"ability": "math", "input": "余弦是什么意思？"},
    config={"configurable": {"session_id": "foobar"}},
)

content='余弦是一个三角函数，它表示直角三角形的邻边长度和斜边长度的比值。' response_metadata={'token_usage': {'completion_tokens': 33, 'prompt_tokens': 38, 'total_tokens': 71}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-2d1eba02-4709-4db5-ab6b-0fd03ab4c68a-0' usage_metadata={'input_tokens': 38, 'output_tokens': 33, 'total_tokens': 71}

with_message_history.invoke(
    {"ability": "math", "input": "什么?"},
    config={"configurable": {"session_id": "foobar"}},
)

content='余弦是一个数学术语，代表在一个角度下的邻边和斜边的比例。' response_metadata={'token_usage': {'completion_tokens': 32, 'prompt_tokens': 83, 'total_tokens': 115}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-99368d03-c2ed-4dda-a32f-677c036ad676-0' usage_metadata={'input_tokens': 83, 'output_tokens': 32, 'total_tokens': 115}

redis历史记录查询
在这里插入图片描述

Track token usage, Cache model responses

Track token usage(跟踪token使用情况)

跟踪令牌使用情况以计算成本是将您的应用投入生产的重要部分。本指南介绍了如何从您的 LangChain 模型调用中获取此信息。

使用 AIMessage.response_metadata

许多模型提供程序将令牌使用信息作为聊天生成响应的一部分返回。如果可用，这将包含在 AIMessage.response_metadata 字段中。以下是一个使用 OpenAI 的示例：

# !pip install -qU langchain-openai
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4-turbo")
msg = llm.invoke([("human", "最古老的楔形文字的已知例子是什么")])
msg.response_metadata

{'token_usage': {'completion_tokens': 114, 'prompt_tokens': 25, 'total_tokens': 139}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None}

使用回调

还有一些特定于 API 的回调上下文管理器，允许您跟踪多个调用中的令牌使用情况。目前仅为 OpenAI API 和 Bedrock Anthropic API 实现了此功能。
让我们首先看一个极其简单的示例，用于跟踪单个 Chat 模型调用的令牌使用情况。

# !pip install -qU langchain-community wikipedia
from langchain_community.callbacks.manager import get_openai_callback
llm = ChatOpenAI(model="gpt-4", temperature=0)
with get_openai_callback() as cb:
    result = llm.invoke("告诉我一个笑话")
    print(cb)

Tokens Used: 59
Prompt Tokens: 14
Completion Tokens: 45
Successful Requests: 1
Total Cost (USD): $0.0031199999999999995
----------------------------------------
使用的令牌数：59
    提示令牌：14
    完成令牌：4
成功请求次数：1
总成本（美元）：$0.0031199999999999995

上下文管理器中的任何内容都将被跟踪。以下是在其中使用它来跟踪连续多次调用的示例。

with get_openai_callback() as cb:
    result = llm.invoke("告诉我一个笑话")
        result2 = llm.invoke("告诉我一个笑话")
    print(cb.total_tokens)

如果使用具有多个步骤的链或代理，它将跟踪所有这些步骤。

from langchain.agents import AgentExecutor, create_tool_calling_agent, load_tools
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "您是一个乐于助人的助手"),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ]
)
# 可能需要科学上网（也就是梯子）
tools = load_tools(["wikipedia"])
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent, tools=tools, verbose=True, stream_runnable=False
)

我们必须将 stream_runnable=False 设置为令牌计数才能正常工作。默认情况下，AgentExecutor 将流式传输底层代理，以便在通过 AgentExecutor.stream_events 流式传输事件时获得最精细的结果。但是，OpenAI 在流式传输模型响应时不会返回令牌计数，因此我们需要关闭底层流式传输。

with get_openai_callback() as cb:
    response = agent_executor.invoke(
        {
            "input": "蜂鸟的学名是什么，哪种鸟是最快的？"
        }
    )
    print(f"总令牌数：{cb.total_tokens}")
    print(f"提示令牌：{cb.prompt_tokens}")
    print(f"完成令牌：{cb.completion_tokens}")
    print(f"总成本（美元）：${cb.total_cost}")

> Entering new AgentExecutor chain...

Invoking: `wikipedia` with `{'query': '蜂鸟'}`


Page: Hawick Lau
Summary: Hawick Lau Hoi-wai (Chinese: 劉愷威; born 13 October 1974) is a Hong Kong actor and singer. He was named as one of the "Five Fresh Tigers of TVB" and is best known for his performances in the series A Kindred Spirit (1995), Virtues of Harmony (2001) and My Family (2005).
He then expanded his career into mainland China, acting in several notable series. His notable appearances include Sealed with a Kiss (2011), A Clear Midsummer Night (2013), The Wife's Secret (2014), Lady & Liar (2015) and Chronicle of Life (2016).

Page: Zhang Jianing
Summary: Zhang Jianing (Chinese: 张佳宁, born 26 May 1989), also known as Karlina Zhang, is a Chinese actress. She is best known for her roles as Muyun Yanshuang in Tribes and Empires: Storm of Prophecy (2017) and Lin Beixing in Shining for One Thing (2022).

Page: Li Xirui
Summary: Li Xirui (Chinese: 李溪芮; born 30 January 1990) is a Chinese actress and singer.
Invoking: `wikipedia` with `{'query': '蜂鸟学名'}`


No good Wikipedia Search Result was found
Invoking: `wikipedia` with `{'query': 'fastest bird'}`


Page: Fastest animals
Summary: This is a list of the fastest animals in the world, by types of animal.

Page: List of birds by flight speed
Summary: This is a list of the fastest flying birds in the world. A bird's velocity is necessarily variable; a hunting bird will reach much greater speeds while diving to catch prey than when flying horizontally. The bird that can achieve the greatest airspeed is the peregrine falcon (Falco peregrinus), able to exceed 320 km/h (200 mph) in its dives. A close relative of the common swift, the white-throated needletail (Hirundapus caudacutus), is commonly reported as the fastest bird in level flight with a reported top speed of 169 km/h (105 mph). This record remains unconfirmed as the measurement methods have never been published or verified. The record for the fastest confirmed level flight by a bird is 111.5 km/h (69.3 mph) held by the common swift.

Page: Abdul Khaliq (athlete)
Summary: Subedar Abdul Khaliq (Punjabi, Urdu: عبد الخالق; 23 March 1933 – 10 March 1988), also known as Parinda-e-Asia (Urdu for The Flying Bird of Asia), was a Pakistani sprinter from 8 Medium Regiment Artillery who won 36 international gold medals, 15 international silver medals, and 12 International bronze medals while representing Pakistan. He competed in the 100m, 200m, and 4 x 100 meters relay. He participated in the 1956 Melbourne Olympics and the 1960 Rome Olympics. He also participated in the 1954 Asian Games and the 1958 Asian Games. During the 1956 Indo-Pak Meet held in Delhi, Abdul Khaliq was first referred to as "The Flying Bird of Asia" by the Prime Minister of India of the time was Jawaharlal Nehru, who was reportedly captivated by his performance during the event.蜂鸟的学名是Trochilidae。最快的鸟是游隼（Falco peregrinus），在俯冲捕食时，速度可以超过320公里/小时（200英里/小时）。在水平飞行中，最快的鸟是普通雨燕，其确认的最高速度为111.5公里/小时（69.3英里/小时）。

> Finished chain.
总令牌数：2088
提示令牌：1922
完成令牌：166
总成本（美元）：$0.06762

Cache model responses

LangChain为聊天模型提供了一个可选的缓存层。这很有用，主要有两个原因：
● 如果您经常多次请求相同的完成，它可以通过减少您向 LLM 提供商进行的 API 调用次数来为您节省资金。这在应用程序开发过程中特别有用。
● 它可以通过减少您向 LLM 提供商进行的 API 调用次数来加快您的应用程序。

pip install -qU langchain-openai

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o")

from langchain.globals import set_llm_cache

API Reference: set_llm_cache

In Memory Cache 内存缓存中

This is an ephemeral cache that stores model calls in memory. It will be wiped when your environment restarts, and is not shared across processes.
这是一个临时缓存，用于在内存中存储模型调用。当您的环境重新启动时，它将被擦除，并且不会在进程之间共享。

from langchain.globals import set_llm_cache
from langchain_community.cache import InMemoryCache

# 创建LLM实例
llm = ChatOpenAI(model="gpt-4")
set_llm_cache(InMemoryCache())

def measure_invoke_time(llm, prompt):
    # 记录开始时间
    start_wall_time = time.time()
    start_cpu_times = os.times()

    # 调用LLM
    response = llm.invoke(prompt)

    # 记录结束时间
    end_wall_time = time.time()
    end_cpu_times = os.times()

    # 计算经过的时间
    wall_time = end_wall_time - start_wall_time
    user_time = end_cpu_times.user - start_cpu_times.user
    sys_time = end_cpu_times.system - start_cpu_times.system
    total_cpu_time = user_time + sys_time
    return response, wall_time, user_time, sys_time, total_cpu_time

API Reference: InMemoryCache

# 第一次调用
First call response: content='当然，这是一则关于数学的笑话：\n\n为什么植物恨数学？\n\n因为它给他们太多的根问题（sqrt问题）。' response_metadata={'token_usage': {'completion_tokens': 46, 'prompt_tokens': 14, 'total_tokens': 60}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-40d86131-39ad-42c9-b3b8-5a422343ba9b-0' usage_metadata={'input_tokens': 14, 'output_tokens': 46, 'total_tokens': 60}
First call CPU times: user 109 ms, sys: 31 ms, total: 141 ms
First call Wall time: 3654 ms

content='当然，这是一则关于数学的笑话：\n\n为什么植物恨数学？\n\n因为它给他们太多的根问题（sqrt问题）。'

#第二次调用使用缓存，所以速度很快
response2, wall_time2, user_time2, sys_time2, total_cpu_time2 = measure_invoke_time(llm, "给我讲个笑话")
print("Second call response:", response2)
print(f"Second call CPU times: user {user_time2 * 1000:.0f} ms, sys: {sys_time2 * 1000:.0f} ms, total: {total_cpu_time2 * 1000:.0f} ms")
print(f"Second call Wall time: {wall_time2 * 1000:.0f} ms")

Second call response: content='当然，这是一则关于数学的笑话：\n\n为什么植物恨数学？\n\n因为它给他们太多的根问题（sqrt问题）。' response_metadata={'token_usage': {'completion_tokens': 46, 'prompt_tokens': 14, 'total_tokens': 60}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-40d86131-39ad-42c9-b3b8-5a422343ba9b-0' usage_metadata={'input_tokens': 14, 'output_tokens': 46, 'total_tokens': 60}
Second call CPU times: user 16 ms, sys: 0 ms, total: 16 ms
Second call Wall time: 1 ms

QLite Cache SQLite 缓存

This cache implementation uses a SQLite database to store responses, and will last across process restarts.
此缓存实现使用 SQLite 数据库来存储响应，并将在进程重启后持续进行。
直接运行生成如下图的文件
在这里插入图片描述

# We can do the same thing with a SQLite cache
from langchain_community.cache import SQLiteCache
set_llm_cache(SQLiteCache(database_path=".langchain.db"))

API Reference: SQLiteCache

# 第一次调用
First call response: content='当然，这是一则关于数学的笑话：\n\n为什么植物恨数学？\n\n因为它给他们太多的根问题（sqrt问题）。' response_metadata={'token_usage': {'completion_tokens': 46, 'prompt_tokens': 14, 'total_tokens': 60}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-40d86131-39ad-42c9-b3b8-5a422343ba9b-0' usage_metadata={'input_tokens': 14, 'output_tokens': 46, 'total_tokens': 60}
First call CPU times: user 109 ms, sys: 31 ms, total: 141 ms
First call Wall time: 3654 ms

content='好的，这是一个关于电脑的笑话：\n\n为什么电脑经常感冒？\n\n因为它窗户(Window)太多了。'

#第二次调用使用缓存，所以速度很快
response2, wall_time2, user_time2, sys_time2, total_cpu_time2 = measure_invoke_time(llm, "给我讲个笑话")
print("Second call response:", response2)
print(f"Second call CPU times: user {user_time2 * 1000:.0f} ms, sys: {sys_time2 * 1000:.0f} ms, total: {total_cpu_time2 * 1000:.0f} ms")
print(f"Second call Wall time: {wall_time2 * 1000:.0f} ms")

Second call response: content='当然，这是一则关于数学的笑话：\n\n为什么植物恨数学？\n\n因为它给他们太多的根问题（sqrt问题）。' response_metadata={'token_usage': {'completion_tokens': 46, 'prompt_tokens': 14, 'total_tokens': 60}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-40d86131-39ad-42c9-b3b8-5a422343ba9b-0' usage_metadata={'input_tokens': 14, 'output_tokens': 46, 'total_tokens': 60}
Second call CPU times: user 16 ms, sys: 0 ms, total: 16 ms
Second call Wall time: 1 ms

content='好的，这是一个关于电脑的笑话：\n\n为什么电脑经常感冒？\n\n因为它窗户(Window)太多了。'

LangChain Chat Model学习笔记