基于 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 --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 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}

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_community.chat_models import ChatTongyi
llm = ChatTongyi(model="qwen-turbo")
msg = llm.invoke([("human", "最古老的楔形文字的已知例子是什么")])
msg.response_metadata

{'token_usage': {'completion_tokens': 114, 'prompt_tokens': 25, 'total_tokens': 139}, 'model_name': 'qwen-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)

fTokens 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)

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

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

使用 requests 的 Python 代码：

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

114

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

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