跳到内容

短期记忆 API 参考

  • SummarizationNode

    一个 LangGraph 节点,当消息超过令牌限制时对其进行摘要,并用一条摘要消息替换它们。

  • SummarizationResult

    消息摘要的结果。

  • RunningSummary

    用于存储有关先前摘要信息的对象。

函数

  • summarize_messages

    当消息超过令牌限制时对其进行摘要,并用一条摘要消息替换它们。

SummarizationNode

基类:RunnableCallable

一个 LangGraph 节点,当消息超过令牌限制时对其进行摘要,并用一条摘要消息替换它们。

方法

  • __init__

    一个 LangGraph 节点,当消息超过令牌限制时对其进行摘要,并用一条摘要消息替换它们。

__init__

__init__(
    *,
    model: LanguageModelLike,
    max_tokens: int,
    max_tokens_before_summary: int | None = None,
    max_summary_tokens: int = 256,
    token_counter: TokenCounter = count_tokens_approximately,
    initial_summary_prompt: ChatPromptTemplate = DEFAULT_INITIAL_SUMMARY_PROMPT,
    existing_summary_prompt: ChatPromptTemplate = DEFAULT_EXISTING_SUMMARY_PROMPT,
    final_prompt: ChatPromptTemplate = DEFAULT_FINAL_SUMMARY_PROMPT,
    input_messages_key: str = "messages",
    output_messages_key: str = "summarized_messages",
    name: str = "summarization",
) -> None

一个 LangGraph 节点,当消息超过令牌限制时对其进行摘要,并用一条摘要消息替换它们。

从最旧到最新处理消息:一旦消息的累积令牌数达到 max_tokens_before_summary,所有在 max_tokens_before_summary 范围内的消息(如果存在,则不包括系统消息)都将被摘要,并替换为一条新的摘要消息。最终生成的消息列表为 [摘要消息] + 剩余消息。

参数

  • model (LanguageModelLike) –

    用于生成摘要的语言模型。

  • max_tokens (int) –

    最终输出中返回的最大令牌数。仅在摘要后强制执行。

  • max_tokens_before_summary (int | None, default: None ) –

    在触发摘要前累积的最大令牌数。如果未提供,则默认为与 max_tokens 相同的值。如果需要,这允许将更多令牌放入摘要 LLM 中。

    注意

    如果 max_tokens_before_summary 内的最后一条消息是带有工具调用的 AI 消息,则所有后续对应的工具消息也将被摘要。

    注意

    如果待摘要的令牌数大于 max_tokens,则仅对其中最后的 max_tokens 进行摘要。这样做是为了防止超出摘要 LLM 的上下文窗口(假定其上限为 max_tokens)。

  • max_summary_tokens (int, default: 256 ) –

    为摘要预算的最大令牌数。

    注意

    此参数不会传递给生成摘要的 LLM 以限制摘要的长度。它仅用于正确估算最大允许的令牌预算。如果要强制执行此限制,您需要将 model.bind(max_tokens=max_summary_tokens) 作为 model 参数传递给此函数。

  • token_counter (TokenCounter, default: count_tokens_approximately ) –

    用于计算消息中令牌数的函数。默认为近似计数。要获得更准确的计数,您可以使用 model.get_num_tokens_from_messages

  • initial_summary_prompt (ChatPromptTemplate, default: DEFAULT_INITIAL_SUMMARY_PROMPT ) –

    用于生成第一个摘要的提示模板。

  • existing_summary_prompt (ChatPromptTemplate, default: DEFAULT_EXISTING_SUMMARY_PROMPT ) –

    用于更新现有(运行中)摘要的提示模板。

  • final_prompt (ChatPromptTemplate, default: DEFAULT_FINAL_SUMMARY_PROMPT ) –

    在返回前将摘要与剩余消息结合的提示模板。

  • input_messages_key (str, default: 'messages' ) –

    输入图状态中包含待摘要消息列表的键。

  • output_messages_key (str, default: 'summarized_messages' ) –

    状态更新中包含更新后消息列表的键。

    警告

    默认情况下,output_messages_keyinput_messages_key 不同。这样做是为了将摘要后的消息与图状态中的主消息列表(即 input_messages_key)解耦。只有当您想覆盖主消息列表(即 input_messages_key)时,才应将它们设为相同。

  • name (str, default: 'summarization' ) –

    摘要节点的名称。

返回

  • None

    以下格式的 LangGraph 状态更新

    {
        "output_messages_key": <list of updated messages ready to be input to the LLM after summarization, including a message with a summary (if any)>,
        "context": {"running_summary": <RunningSummary object>}
    }
    

示例
from typing import Any, TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import AnyMessage
from langgraph.graph import StateGraph, START, MessagesState
from langgraph.checkpoint.memory import InMemorySaver
from langmem.short_term import SummarizationNode, RunningSummary

model = ChatOpenAI(model="gpt-4o")
summarization_model = model.bind(max_tokens=128)


class State(MessagesState):
    context: dict[str, Any]


class LLMInputState(TypedDict):
    summarized_messages: list[AnyMessage]
    context: dict[str, Any]


summarization_node = SummarizationNode(
    model=summarization_model,
    max_tokens=256,
    max_tokens_before_summary=256,
    max_summary_tokens=128,
)


def call_model(state: LLMInputState):
    response = model.invoke(state["summarized_messages"])
    return {"messages": [response]}


checkpointer = InMemorySaver()
workflow = StateGraph(State)
workflow.add_node(call_model)
workflow.add_node("summarize", summarization_node)
workflow.add_edge(START, "summarize")
workflow.add_edge("summarize", "call_model")
graph = workflow.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "1"}}
graph.invoke({"messages": "hi, my name is bob"}, config)
graph.invoke({"messages": "write a short poem about cats"}, config)
graph.invoke({"messages": "now do the same but for dogs"}, config)
graph.invoke({"messages": "what's my name?"}, config)

SummarizationResult dataclass

消息摘要的结果。

属性

  • messages (list[AnyMessage]) –

    摘要后准备输入 LLM 的更新后消息列表,包括带有摘要的消息(如果有的话)。

  • running_summary (RunningSummary | None) –

    关于先前摘要的信息(摘要和先前摘要消息的 ID)。

messages instance-attribute

messages: list[AnyMessage]

摘要后准备输入 LLM 的更新后消息列表,包括带有摘要的消息(如果有的话)。

running_summary class-attribute instance-attribute

running_summary: RunningSummary | None = None

关于先前摘要的信息(摘要和先前摘要消息的 ID)。如果未执行摘要(消息不足以进行摘要),则可以为 None。

RunningSummary dataclass

用于存储有关先前摘要信息的对象。

在后续调用 summarize_messages 时使用,以避免对相同的消息进行摘要。

属性

summary instance-attribute

summary: str

消息的最新摘要,每次执行摘要时都会更新。

summarized_message_ids instance-attribute

summarized_message_ids: set[str]

所有先前已摘要消息的 ID。

last_summarized_message_id instance-attribute

last_summarized_message_id: str | None

最后一条被摘要的消息的 ID。

summarize_messages

summarize_messages(
    messages: list[AnyMessage],
    *,
    running_summary: RunningSummary | None,
    model: LanguageModelLike,
    max_tokens: int,
    max_tokens_before_summary: int | None = None,
    max_summary_tokens: int = 256,
    token_counter: TokenCounter = count_tokens_approximately,
    initial_summary_prompt: ChatPromptTemplate = DEFAULT_INITIAL_SUMMARY_PROMPT,
    existing_summary_prompt: ChatPromptTemplate = DEFAULT_EXISTING_SUMMARY_PROMPT,
    final_prompt: ChatPromptTemplate = DEFAULT_FINAL_SUMMARY_PROMPT,
) -> SummarizationResult

当消息超过令牌限制时对其进行摘要,并用一条摘要消息替换它们。

此函数从最旧到最新处理消息:一旦消息的累积令牌数达到 max_tokens_before_summary,所有在 max_tokens_before_summary 范围内的消息(如果存在,则不包括系统消息)都将被摘要,并替换为一条新的摘要消息。最终生成的消息列表为 [摘要消息] + 剩余消息。

参数

  • messages (list[AnyMessage]) –

    要处理的消息列表。

  • running_summary (RunningSummary | None) –

    可选的运行中摘要对象,包含有关先前摘要的信息。如果提供:- 仅处理先前被摘要的消息 - 如果未生成新摘要,则将运行中摘要添加到返回的消息中 - 如果需要生成新摘要,则通过合并运行中摘要中的现有摘要值来生成

  • model (LanguageModelLike) –

    用于生成摘要的语言模型。

  • max_tokens (int) –

    最终输出中返回的最大令牌数。仅在摘要后强制执行。这也将用作提供给摘要 LLM 的最大令牌数。

  • max_tokens_before_summary (int | None, default: None ) –

    在触发摘要前累积的最大令牌数。如果未提供,则默认为与 max_tokens 相同的值。如果需要,这允许将更多令牌放入摘要 LLM 中。

    注意

    如果 max_tokens_before_summary 内的最后一条消息是带有工具调用的 AI 消息,则所有后续对应的工具消息也将被摘要。

    注意

    如果待摘要的令牌数大于 max_tokens,则仅对其中最后的 max_tokens 进行摘要。这样做是为了防止超出摘要 LLM 的上下文窗口(假定其上限为 max_tokens)。

  • max_summary_tokens (int, default: 256 ) –

    为摘要预算的最大令牌数。

    注意

    此参数不会传递给生成摘要的 LLM 以限制摘要的长度。它仅用于正确估算最大允许的令牌预算。如果要强制执行此限制,您需要将 model.bind(max_tokens=max_summary_tokens) 作为 model 参数传递给此函数。

  • token_counter (TokenCounter, default: count_tokens_approximately ) –

    用于计算消息中令牌数的函数。默认为近似计数。要获得更准确的计数,您可以使用 model.get_num_tokens_from_messages

  • initial_summary_prompt (ChatPromptTemplate, default: DEFAULT_INITIAL_SUMMARY_PROMPT ) –

    用于生成第一个摘要的提示模板。

  • existing_summary_prompt (ChatPromptTemplate, default: DEFAULT_EXISTING_SUMMARY_PROMPT ) –

    用于更新现有(运行中)摘要的提示模板。

  • final_prompt (ChatPromptTemplate, default: DEFAULT_FINAL_SUMMARY_PROMPT ) –

    在返回前将摘要与剩余消息结合的提示模板。

返回

  • SummarizationResult

    一个 SummarizationResult 对象,包含更新后的消息和运行中摘要。- messages: 准备输入 LLM 的更新后消息列表 - running_summary: RunningSummary 对象 - summary: 最新摘要的文本 - summarized_message_ids: 先前已摘要消息的 ID 集合 - last_summarized_message_id: 最后一条被摘要的消息的 ID

示例
from langgraph.graph import StateGraph, START, MessagesState
from langgraph.checkpoint.memory import InMemorySaver
from langmem.short_term import summarize_messages, RunningSummary
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")
summarization_model = model.bind(max_tokens=128)


class SummaryState(MessagesState):
    summary: RunningSummary | None


def call_model(state):
    summarization_result = summarize_messages(
        state["messages"],
        running_summary=state.get("summary"),
        model=summarization_model,
        max_tokens=256,
        max_tokens_before_summary=256,
        max_summary_tokens=128,
    )
    response = model.invoke(summarization_result.messages)
    state_update = {"messages": [response]}
    if summarization_result.running_summary:
        state_update["summary"] = summarization_result.running_summary
    return state_update


checkpointer = InMemorySaver()
workflow = StateGraph(SummaryState)
workflow.add_node(call_model)
workflow.add_edge(START, "call_model")
graph = workflow.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "1"}}
graph.invoke({"messages": "hi, my name is bob"}, config)
graph.invoke({"messages": "write a short poem about cats"}, config)
graph.invoke({"messages": "now do the same but for dogs"}, config)
graph.invoke({"messages": "what's my name?"}, config)

评论