Prompt templates: Few shot、Example selector

Few shot(少量示例)

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

创建一个简单的提示模板，用于在生成时向模型提供示例输入和输出。向 LLM 提供少量这样的示例被称为少量示例，这是一种简单但强大的指导生成的方式，在某些情况下可以显著提高模型性能。

少量示例提示模板可以由一组示例或一个负责从定义的集合中选择一部分示例的示例选择器类构建。

配置一个格式化程序，将少量示例格式化为字符串。这个格式化程序应该是一个 PromptTemplate 对象。

 #技术分享from langchain_core.prompts import PromptTemplateexample_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 FewShotPromptTemplateprompt = 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日 ​ ​ 问题：乔治·华盛顿的外祖父是谁？​ 是否需要后续问题：是的。后续问题：乔治·华盛顿的母亲是谁？中间答案：乔治·华盛顿的母亲是玛丽·波尔·华盛顿。后续问题：玛丽·波尔·华盛顿的父亲是谁？中间答案：玛丽·波尔·华盛顿的父亲是约瑟夫·波尔。所以最终答案是：约瑟夫·波尔 ​ ​ 问题：《大白鲨》和《皇家赌场》的导演都来自同一个国家吗？​ 是否需要后续问题：是的。后续问题：《大白鲨》的导演是谁？中间答案：《大白鲨》的导演是史蒂文·斯皮尔伯格。后续问题：史蒂文·斯皮尔伯格来自哪个国家？中间答案：美国。后续问题：《皇家赌场》的导演是谁？中间答案：《皇家赌场》的导演是马丁·坎贝尔。后续问题：马丁·坎贝尔来自哪个国家？中间答案：新西兰。所以最终答案是：不是 ​ ​ 问题：乔治·华盛顿的父亲是谁？

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

C Example selectors(示例选择器)

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

为了展示它的样子，让我们初始化一个实例并在隔离环境中调用它：

from langchain_chroma import Chromafrom langchain_core.example_selectors import SemanticSimilarityExampleSelectorfrom langchain_openai import OpenAIEmbeddingsexample_selector = SemanticSimilarityExampleSelector.from_examples(        examples,        OpenAIEmbeddings(),        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}")

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

Chroma 安装报错

pip install langchain-chromaerror: Microsoft Visual C++[end of output]

解决方案：需要先下载 visual-cpp-build-tools，再执行 pip install langchain-chroma

现在，让我们创建一个 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）

限制

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

安装

对于客户端和服务器：

pip install --upgrade "langserve[all]"

或者对于客户端代码，pip install "langserve[client]" ，对于服务器代码，pip install "langserve[server]" 。

C LangChain CLI ️

使用 LangChain CLI 快速启动 LangServe 项目。

要使用 langchain CLI，请确保已安装最新版本的 langchain-cli 。您可以使用 pip install -U langchain-cli 进行安装。

设置

注意 ：我们使用 poetry 进行依赖管理。请参阅 poetry 文档了解更多信息。

1. 使用 langchain cli 命令创建新应用

langchain app new my-app

2. 在 add_routes 中定义可运行对象。转到 server.py 并编辑

add_routes(app. NotImplemented)

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

#安装pipx，参考：https://pipx.pypa.io/stable/installation/pip install pipx#加入到环境变量，需要重启PyCharmpipx ensurepath# 安装poetry，参考：https://python-poetry.org/docs/pipx install poetry   #安装 langchain-openai 库，例如：poetry add [package-name] poetry add langchain poetry add langchain-openai

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

export OPENAI_API_KEY="sk-..."

5. 启动您的应用

poetry run langchain serve --port=8000

示例应用

服务器

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

from fastapi import FastAPIfrom langchain_openai import ChatOpenAIfrom langserve import add_routesapp = 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 CORSMiddlewareapp.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 RunnableMapfrom langchain_core.prompts import ChatPromptTemplatefrom langserve import RemoteRunnableopenai = 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 requestsresponse = 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

C 对话状态Chain传递

在构建聊天机器人时，将对话状态传递到链中以及从链中传出对话状态至关重要。RunnableWithMessageHistory 类让我们能够向某些类型的链中添加消息历史。它包装另一个 Runnable 并管理其聊天消息历史。

具体来说，它可用于任何接受以下之一作为输入的 Runnable：

一系列 BaseMessages 具有以序列 BaseMessages 作为值的键的字典并将以下之一作为输出返回：

• 可视为 AIMessage 内容的字符串
• 一系列 BaseMessage
• 具有包含一系列 BaseMessage 的键的字典

让我们看一些示例以了解其工作原理。首先，我们构建一个 Runnable（此处接受字典作为输入并返回消息作为输出）：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderfrom langchain_openai.chat_models import ChatOpenAIprompt = 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='')

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

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

聊天历史存储在内存

下面我们展示一个简单的示例，其中聊天历史保存在内存中，此处通过全局 Python 字典实现。

我们构建一个名为 get_session_history 的可调用对象，引用此字典以返回 ChatMessageHistory 实例。通过在运行时向 RunnableWithMessageHistory 传递配置，可以指定可调用对象的参数。默认情况下，期望配置参数是一个字符串 session_id 。可以通过 history_factory_config 关键字参数进行调整。

使用单参数默认值：

from langchain_community.chat_message_histories import ChatMessageHistoryfrom langchain_core.chat_history import BaseChatMessageHistoryfrom langchain_core.runnables.history import RunnableWithMessageHistorystore = {}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"}},)

content='余弦是一个数学函数，通常在三角学中使用，表示直角三角形的邻边和斜边的比例。' response_metadata={'token_usage': {'completion_tokens': 38, 'prompt_tokens': 38, 'total_tokens': 76}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-9aa23716-3959-476d-9386-6d433266e060-0' usage_metadata={'input_tokens': 38, 'output_tokens': 38, 'total_tokens': 76}

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

content='余弦是一个数学术语，用于描述直角三角形中的角度关系。' response_metadata={'token_usage': {'completion_tokens': 26, 'prompt_tokens': 88, 'total_tokens': 114}, 'model_name': 'gpt-4-0613', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-f77baf90-6a13-4f48-991a-28e60ece84e8-0' usage_metadata={'input_tokens': 88, 'output_tokens': 26, 'total_tokens': 114}

# 新的 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 ConfigurableFieldSpecstore = {}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 --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是否存储历史记录

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

from langchain_community.chat_message_histories import RedisChatMessageHistorydef 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 的示例：

from langchain_openai import ChatOpenAIllm = 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 wikipediafrom 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: 59Prompt Tokens: 14Completion Tokens: 45Successful Requests: 1Total 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)

114

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

from langchain.agents import AgentExecutor, create_tool_calling_agent, load_toolsfrom langchain_core.prompts import ChatPromptTemplateprompt = 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 `{> > > 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> > 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 `{> > > No good Wikipedia Search Result was found> Invoking: `wikipedia` with `{> > > 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> > 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

C Cache model responses

LangChain 为聊天模型提供了一个可选的缓存层。这很有用，主要有两个原因：

• 如果您经常多次请求相同的完成，它可以通过减少您向 LLM 提供商进行的 API 调用次数来为您节省资金。这在应用程序开发过程中特别有用。
• 它可以通过减少您向 LLM 提供商进行的 API 调用次数来加快您的应用程序。

pip install -qU langchain-openai

from langchain_openai import ChatOpenAIllm = 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_cachefrom langchain_community.cache import InMemoryCachellm = ChatOpenAI(model="gpt-4") set_llm_cache(InMemoryCache())  def measure_invoke_time(llm, prompt): start_wall_time = time.time() start_cpu_times = os.times()  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 *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 msSecond call Wall time: 1 ms

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

SQLite Cache SQLite 缓存

This cache implementation uses a SQLite database to store responses, and will last across process restarts. 此缓存实现使用 SQLite 数据库来存储响应，并将在进程重启后持续进行。

from langchain_community.cache import SQLiteCacheset_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 *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 msSecond call Wall time: 1 ms

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