会话记忆与多轮上下文

# -*- coding: utf-8 -*-
"""
消息与历史容器：三类消息 + InMemoryChatMessageHistory
=========================================================
会话历史的最小单位是「消息对象」，不是字符串。
把消息拼成一段纯文本很容易丢角色信息，多轮之后模型就分不清谁说了什么。

InMemoryChatMessageHistory 是最基础的历史容器：
只负责存消息、取消息、清空，不做任何裁剪、摘要或格式化。
"""

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage


def build_history():
    """手工塞入一段三轮对话，观察它在内存里长什么样。"""
    history = InMemoryChatMessageHistory()

    # 方式一：便捷方法，内部帮你包成对应的消息类
    history.add_user_message("你好，我叫小张")
    history.add_ai_message("你好小张，很高兴认识你")

    # 方式二：直接追加消息对象，能一次加多条、也能加 SystemMessage
    history.add_messages([
        HumanMessage(content="帮我记一下：我的项目叫「中央厨房」"),
        AIMessage(content="好的，已经记住你的项目名是「中央厨房」"),
    ])
    return history


def inspect(history):
    """messages 属性返回 List[BaseMessage]，可以按下标、按类型访问。"""
    for i, msg in enumerate(history.messages, 1):
        # type 是 'human' / 'ai' / 'system' / 'tool'
        print("%d. [%s] %s" % (i, msg.type, msg.content))

    print("总条数：", len(history.messages))
    print("只看用户说过的话：",
          [m.content for m in history.messages if isinstance(m, HumanMessage)])


def system_message_is_not_history():
    """SystemMessage 是人设，不属于会话历史，每轮由提示词模板重新拼上。

    把它存进历史容器，一旦历史被裁剪，人设就可能被一起裁掉。
    正确做法是让它固定待在提示词模板的第一条。
    """
    persona = SystemMessage(content="你是一个耐心的中文助教，回答控制在三句话以内。")
    history = build_history()
    return [persona] + history.messages


if __name__ == "__main__":
    h = build_history()
    inspect(h)

    # 清空整段会话：用户点「新建对话」时就是调它
    h.clear()
    print("清空后：", h.messages)          # []

    print("拼上人设后共 %d 条" % len(system_message_is_not_history()))

# -*- coding: utf-8 -*-
"""
消息与 dict 的互转，以及把会话落盘
===================================
进程一退出，内存里的历史就没了。想让「昨天聊到一半」明天还能接上，
就得把消息对象转成可序列化的结构存起来，下次再转回来。

core 提供了一对函数：
    messages_to_dict(msgs)    消息对象 -> 可 json.dumps 的 list[dict]
    messages_from_dict(dicts) list[dict] -> 消息对象
"""

import json
import os

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import messages_from_dict, messages_to_dict

STORE_DIR = os.environ.get("CHAT_STORE_DIR", "./chat_sessions")


def dump_session(session_id, history):
    """把一段会话写成 JSON 文件。"""
    os.makedirs(STORE_DIR, exist_ok=True)
    path = os.path.join(STORE_DIR, "%s.json" % session_id)

    payload = messages_to_dict(history.messages)
    # payload 形如：
    # [{"type": "human", "data": {"content": "在吗", "additional_kwargs": {}, ...}},
    #  {"type": "ai",    "data": {"content": "有什么事？", ...}}]

    with open(path, "w", encoding="utf-8") as f:
        # ensure_ascii=False：中文按原样写入，文件打开就能读
        json.dump(payload, f, ensure_ascii=False, indent=2)
    return path


def load_session(session_id):
    """把 JSON 文件读回成一个可继续追加的历史容器。"""
    path = os.path.join(STORE_DIR, "%s.json" % session_id)
    history = InMemoryChatMessageHistory()
    if not os.path.exists(path):
        return history            # 新会话，返回空容器

    with open(path, encoding="utf-8") as f:
        payload = json.load(f)

    # 转回来的是 HumanMessage / AIMessage 对象，不是 dict
    history.add_messages(messages_from_dict(payload))
    return history


if __name__ == "__main__":
    h = InMemoryChatMessageHistory()
    h.add_user_message("在吗")
    h.add_ai_message("有什么事？")

    p = dump_session("u1001", h)
    print("已写入", p)

    back = load_session("u1001")
    print("读回 %d 条" % len(back.messages))
    print("第一条类型：", type(back.messages[0]).__name__)   # HumanMessage
    print("内容一致：", back.messages[0].content == "在吗")   # True

    # 接着聊，新消息追加在后面，再 dump 一次即可
    back.add_user_message("帮我查下订单")
    dump_session("u1001", back)

# -*- coding: utf-8 -*-
"""
旧写法与当代写法的迁移
========================
LangChain 0.3 时代的 Memory 是一组「记忆类」：ConversationBufferMemory、
ConversationBufferWindowMemory、ConversationSummaryMemory、
ConversationSummaryBufferMemory，以及把它们和 LLMChain 封装到一起的
ConversationChain。

在 1.x 里，这些类连同 LLMChain 一起搬到了 langchain-classic 包，
不再是推荐写法：它们接不进 LCEL 管道，也没法用 | 往后接输出解析器。
当代做法是「链 + RunnableWithMessageHistory」，策略由你在读历史那一步决定。

本文件把旧写法注释在上、新写法写在下，一行对一行地看。
"""

from langchain.chat_models import init_chat_model
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import trim_messages
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough
from langchain_core.runnables.history import RunnableWithMessageHistory

model = init_chat_model("openai:gpt-4o-mini", temperature=0)

# ---------------------------------------------------------------------------
# 旧：memory = ConversationBufferMemory(return_messages=True)
#     chain  = LLMChain(llm=llm, prompt=prompt, memory=memory)
#     chain.invoke({"question": "..."})
# 新：
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个与人类对话的机器人。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "问题：{question}"),
])

_store = {}


def get_history(session_id):
    return _store.setdefault(session_id, InMemoryChatMessageHistory())


# 新写法能接输出解析器，旧的 LLMChain 接不了
buffer_chain = RunnableWithMessageHistory(
    prompt | model | StrOutputParser(),
    get_history,
    input_messages_key="question",
    history_messages_key="history",
)

# ---------------------------------------------------------------------------
# 旧：ConversationBufferWindowMemory(k=2)
# 新：读历史时切片，或用 trim_messages 按 token 裁
window_chain = RunnableWithMessageHistory(
    RunnablePassthrough.assign(history=lambda x: x["history"][-4:])
    | prompt | model | StrOutputParser(),
    get_history, input_messages_key="question", history_messages_key="history",
)

# ---------------------------------------------------------------------------
# 旧：ConversationTokenBufferMemory(llm=llm, max_token_limit=50)
# 新：trim_messages，token_counter 传模型，口径和计费一致
token_chain = RunnableWithMessageHistory(
    RunnablePassthrough.assign(
        history=lambda x: trim_messages(x["history"], max_tokens=200,
                                        strategy="last", token_counter=model,
                                        include_system=True, start_on="human"))
    | prompt | model | StrOutputParser(),
    get_history, input_messages_key="question", history_messages_key="history",
)

# ---------------------------------------------------------------------------
# 旧：ConversationSummaryMemory / ConversationSummaryBufferMemory
# 新：自己维护一句摘要，见 strategy_summary.py 与 strategy_summary_window.py
#
# 旧：ConversationChain(llm=llm)（内置 input / history 两个变量）
# 新：上面的 buffer_chain 就是它的等价物，而且提示词完全由你掌握
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    cfg = {"configurable": {"session_id": "migrate-demo"}}
    print(buffer_chain.invoke({"question": "小明有 1 只猫"}, config=cfg))
    print(buffer_chain.invoke({"question": "小刚有 2 只狗"}, config=cfg))
    print(buffer_chain.invoke({"question": "他们一共有几只宠物？"}, config=cfg))

# -*- coding: utf-8 -*-
"""
Agent 那一侧的短期记忆：checkpointer + thread_id
==================================================
链路上用 RunnableWithMessageHistory 按 session_id 取历史；
Agent 上则是另一套：给 create_agent 传一个 checkpointer，
再用 thread_id 区分会话。

状态存在 graph state 里，由 checkpointer 负责持久化；
thread 之间互相隔离，每一步开始时读、某一步完成时写。

这里只演示形态，Agent 的构建、工具与中间件在 Agent 实战那一讲展开。
"""

from langchain.agents import create_agent
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver


@tool
def get_order_status(order_id: str) -> str:
    """根据订单号查询订单的当前状态。"""
    return "订单 %s 已发货，预计明天送达。" % order_id


# checkpointer 就是 Agent 的备忘板；换成持久化实现即可跨进程保留
agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[get_order_status],
    system_prompt="你是电商客服助手，用中文作答。",
    checkpointer=InMemorySaver(),
)

if __name__ == "__main__":
    cfg = {"configurable": {"thread_id": "u1001-c1"}}

    r1 = agent.invoke(
        {"messages": [{"role": "user", "content": "帮我查一下订单 12346"}]}, cfg)
    print(r1["messages"][-1].content)

    # 同一个 thread_id：不用再说一遍订单号
    r2 = agent.invoke(
        {"messages": [{"role": "user", "content": "它什么时候到？"}]}, cfg)
    print(r2["messages"][-1].content)

    # 换 thread_id 等于换一块空白备忘板
    other = {"configurable": {"thread_id": "u1001-c2"}}
    r3 = agent.invoke(
        {"messages": [{"role": "user", "content": "它什么时候到？"}]}, other)
    print(r3["messages"][-1].content)

# -*- coding: utf-8 -*-
"""
证明模型没有记忆：同一个问题，带历史与不带历史的两种结果
=============================================================
跑完这一个文件，「模型本身没有记忆」就不再是一句口号。

两次实验用同一个模型、同一句追问：
  实验 A：每次只发当前这一句 —— 模型答不出名字
  实验 B：把前几轮原样重新发一遍 —— 模型答得出名字

差别不在模型，在你发过去的 messages 列表。
"""

import os

from langchain.chat_models import init_chat_model

# 模型通过统一入口创建；换 provider 只改这一行的字符串
# 需要密钥的 provider 一律从环境变量读，不要写进源码
os.environ.setdefault("OPENAI_API_KEY", os.environ.get("OPENAI_API_KEY", ""))
model = init_chat_model("openai:gpt-4o-mini", temperature=0)

FIRST = "你好，我叫孙小空，今年考上了一本。"
FOLLOW = "我叫什么名字？"


def without_memory():
    """实验 A：每一轮都是一次全新的请求，模型手上只有当前这一句。"""
    model.invoke([{"role": "user", "content": FIRST}])
    # 注意：上面那次调用的内容没有被任何人保存下来
    reply = model.invoke([{"role": "user", "content": FOLLOW}])
    return reply.content


def with_memory():
    """实验 B：自己维护一个 messages 列表，每轮把全部历史重新发过去。"""
    history = []

    history.append({"role": "user", "content": FIRST})
    first_reply = model.invoke(history)
    # 模型的回答也要追加进去，否则下一轮它看不到自己说过什么
    history.append({"role": "assistant", "content": first_reply.content})

    history.append({"role": "user", "content": FOLLOW})
    second_reply = model.invoke(history)
    history.append({"role": "assistant", "content": second_reply.content})

    print("第二次请求实际发送了 %d 条消息" % (len(history) - 1))
    return second_reply.content


if __name__ == "__main__":
    print("A 无历史：", without_memory())
    # 典型输出：抱歉，我不知道你的名字，你可以告诉我。

    print("B 带历史：", with_memory())
    # 典型输出：你叫孙小空。

    # 结论：模型两次都是同一个模型，能力没变。
    # 变的只是「这次请求里有没有把之前说过的话再带上」。

# -*- coding: utf-8 -*-
"""
最小可跑的带记忆链：RunnableWithMessageHistory + MessagesPlaceholder
=====================================================================
四个零件，缺一不可：

  1. ChatPromptTemplate 里留一个 MessagesPlaceholder，它是历史消息的插槽
  2. 一条普通的 LCEL 链：prompt | model
  3. 一个工厂函数：给 session_id，返回这个会话的历史容器
  4. RunnableWithMessageHistory 把上面三者接起来

调用时用 config={"configurable": {"session_id": "..."}} 指定是谁在说话。
"""

import os

from langchain.chat_models import init_chat_model
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory

model = init_chat_model("openai:gpt-4o-mini", temperature=0)

# 1. 提示词模板：人设固定在最前，历史插槽在中间，本轮输入在最后
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个中文助手，回答简洁明了。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{question}"),
])

# 2. 普通的 LCEL 链，本身完全不知道「记忆」这回事
chain = prompt | model

# 3. 工厂函数：一个 session_id 对应一块独立的历史
_store = {}


def get_history(session_id: str) -> InMemoryChatMessageHistory:
    if session_id not in _store:
        _store[session_id] = InMemoryChatMessageHistory()
    return _store[session_id]


# 4. 包一层，链就有了记忆
#    input_messages_key  ：本轮输入在 invoke 的入参里叫什么
#    history_messages_key：历史要填进模板里哪个插槽
conversation = RunnableWithMessageHistory(
    chain,
    get_history,
    input_messages_key="question",
    history_messages_key="history",
)

if __name__ == "__main__":
    cfg = {"configurable": {"session_id": "demo-001"}}

    r1 = conversation.invoke({"question": "我叫孙小空"}, config=cfg)
    print(r1.content)

    r2 = conversation.invoke({"question": "我叫什么名字？"}, config=cfg)
    print(r2.content)          # 你叫孙小空

    # 换一个 session_id，等于换了一块空白备忘板
    other = {"configurable": {"session_id": "demo-002"}}
    r3 = conversation.invoke({"question": "我叫什么名字？"}, config=other)
    print(r3.content)          # 不知道

    # 读写各发生一次：invoke 前读历史填模板，拿到回复后把这一问一答写回去
    print("demo-001 共 %d 条" % len(get_history("demo-001").messages))   # 4
    print("demo-002 共 %d 条" % len(get_history("demo-002").messages))   # 2

# -*- coding: utf-8 -*-
"""
策略一：全量保存 —— 一条不落地把历史全发回去
================================================
最简单，也最贵。

优点：上下文最完整，模型不会「忘记」任何细节
代价：每一轮的输入 token 都比上一轮更多，费用与延迟随轮数线性上涨；
      聊到一定长度会直接顶穿模型的上下文窗口，请求报错

适用：客服工单、问题诊断这类轮数可控（十几轮以内）且细节不能丢的场景。
"""

from langchain.chat_models import init_chat_model
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory

model = init_chat_model("openai:gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个乐于助人的中文助手。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{question}"),
])

_store = {}


def get_history(session_id):
    """全量策略的工厂函数什么都不做 —— 有多少存多少。"""
    return _store.setdefault(session_id, InMemoryChatMessageHistory())


chat = RunnableWithMessageHistory(
    prompt | model, get_history,
    input_messages_key="question", history_messages_key="history",
)


def run_dialogue(session_id, questions):
    cfg = {"configurable": {"session_id": session_id}}
    for q in questions:
        reply = chat.invoke({"question": q}, config=cfg)
        history_len = len(get_history(session_id).messages)
        # 观察点：每问一句，历史就长 2 条，下一轮的输入也跟着变长
        print("[历史 %2d 条] 用户：%s" % (history_len, q))
        print("             助手：%s\n" % reply.content)


if __name__ == "__main__":
    run_dialogue("buffer-demo", [
        "你好，我是孙小空",
        "我还有两个师弟，猪小戒和沙小僧",
        "我今年考上了一本",
        "我叫什么？我有几个师弟？",     # 全量保存：三条信息都答得出来
    ])

    msgs = get_history("buffer-demo").messages
    print("最终历史 %d 条，字符总数 %d"
          % (len(msgs), sum(len(m.content) for m in msgs)))

# -*- coding: utf-8 -*-
"""
策略二：只留最近 K 轮 —— 滑动窗口
====================================
观察发现，太久远的对话对当前这一问往往没有帮助，
于是只把最近 K 轮塞回去，更早的直接丢掉。

优点：每轮输入长度有上限，费用与延迟稳定，永远不会顶穿上下文窗口
代价：窗口外的信息是真的丢了。第 1 轮说的名字，K=1 时到第 3 轮就再也答不出来

实现要点：裁剪发生在「读历史」这一步，不要去改历史容器本身 ——
容器留全量，读的时候只取一段，这样想换策略随时能换。
"""

from langchain.chat_models import init_chat_model
from langchain_core.chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough
from langchain_core.runnables.history import RunnableWithMessageHistory

K = 2                 # 保留最近 K 轮，一轮 = 一问一答 = 2 条消息

model = init_chat_model("openai:gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个乐于助人的中文助手。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{question}"),
])

_store: dict[str, BaseChatMessageHistory] = {}


def get_history(session_id) -> BaseChatMessageHistory:
    return _store.setdefault(session_id, InMemoryChatMessageHistory())


def window(payload):
    """在进入提示词模板之前，把 history 截成最近 K 轮。"""
    msgs = payload["history"]
    payload["history"] = msgs[-2 * K:] if len(msgs) > 2 * K else msgs
    return payload


# RunnablePassthrough.assign 之后再接模板，裁剪就成了链上的一个普通环节
chain = RunnablePassthrough.assign(history=lambda x: window(x)["history"]) | prompt | model

chat = RunnableWithMessageHistory(
    chain, get_history,
    input_messages_key="question", history_messages_key="history",
)

if __name__ == "__main__":
    cfg = {"configurable": {"session_id": "window-demo"}}
    for q in ["你好，我是孙小空",
              "我还有两个师弟，猪小戒和沙小僧",
              "我今年考上了一本",
              "我叫什么？"]:
        print("用户：", q)
        print("助手：", chat.invoke({"question": q}, config=cfg).content, "\n")

    # K=2 时，最后一问看到的只有第 2、3 轮，第 1 轮的名字已经滑出窗口
    print("容器里仍然存着 %d 条（裁剪只发生在读的时候）"
          % len(get_history("window-demo").messages))

# -*- coding: utf-8 -*-
"""
策略三：摘要压缩 —— 把旧对话交给模型缩写成一段话
====================================================
按条数或 token 硬截断，总会切掉一些仍然有用的信息。
摘要的思路是：不丢内容，只压缩体积 —— 让模型把旧对话缩写成几句话。

优点：几十轮之后输入长度依然可控，关键事实（姓名、订单号、偏好）能保住
代价：① 每次更新摘要都要额外调一次模型，多一份钱、多一份延迟
      ② 摘要是有损的，细节和原话会被抹掉
      ③ 摘要之上再做摘要，误差会一层层累积

实现要点：新摘要 = 旧摘要 + 新增对话，不是每次都从头重写整段历史。
"""

from langchain.chat_models import init_chat_model
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

model = init_chat_model("openai:gpt-4o-mini", temperature=0)

SUMMARY_PROMPT = ChatPromptTemplate.from_messages([
    ("system",
     "你在维护一段对话的滚动摘要。把已有摘要和新增对话合并成一段新的中文摘要，"
     "保留姓名、编号、时间、明确的偏好与承诺；不要编造原文没有的内容；不超过 120 字。"),
    ("human", "已有摘要：\n{summary}\n\n新增对话：\n{new_lines}\n\n新摘要："),
])

CHAT_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "你是一个中文助手。以下是你和用户之前对话的摘要：\n{summary}"),
    MessagesPlaceholder(variable_name="recent"),
    ("human", "{question}"),
])

summary_chain = SUMMARY_PROMPT | model


class SummaryMemory:
    """一段会话 = 一句摘要 + 一个原始历史容器。"""

    def __init__(self):
        self.summary = "（暂无）"
        self.history = InMemoryChatMessageHistory()

    def _render(self, msgs):
        role = {"human": "用户", "ai": "助手"}
        return "\n".join("%s：%s" % (role.get(m.type, m.type), m.content) for m in msgs)

    def update(self, question, answer):
        """把这一轮写进历史，并据此刷新摘要。"""
        new_lines = self._render([HumanMessage(content=question),
                                  AIMessage(content=answer)])
        self.history.add_messages([HumanMessage(content=question),
                                   AIMessage(content=answer)])
        self.summary = summary_chain.invoke(
            {"summary": self.summary, "new_lines": new_lines}).content


def ask(mem: SummaryMemory, question: str) -> str:
    # 纯摘要策略：不带任何原始消息，recent 永远是空列表
    reply = (CHAT_PROMPT | model).invoke(
        {"summary": mem.summary, "recent": [], "question": question})
    mem.update(question, reply.content)
    return reply.content


if __name__ == "__main__":
    mem = SummaryMemory()
    for q in ["你好，我想咨询数据分析课程",
              "我有 Python 基础，但没做过项目",
              "课程时长和学费是多少？",
              "我刚才说我有什么基础来着？"]:
        print("用户：", q)
        print("助手：", ask(mem, q))
        print("当前摘要：", mem.summary, "\n")

    # 原始对话一条没少，只是没有全部发给模型
    print("原始历史 %d 条，摘要 %d 字"
          % (len(mem.history.messages), len(mem.summary)))

# -*- coding: utf-8 -*-
"""
策略四：摘要 + 最近若干轮 —— 混合型，生产里最常用
====================================================
纯摘要丢细节，纯窗口丢远期记忆。混合策略把两者接起来：

    [人设] + [旧对话的摘要] + [最近 K 轮原始消息] + [本轮提问]

最近几轮保持原话，模型答起来精确；再往前的只留摘要，体积可控。
代价是实现比前两种复杂，而且摘要仍然要额外调模型。

触发时机：历史长度超过阈值时，才把「溢出的那一段」并进摘要。
不要每轮都重算摘要，那等于把成本翻倍。
"""

from langchain.chat_models import init_chat_model
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

KEEP_ROUNDS = 2          # 最近保留几轮原始消息
TRIGGER = 6              # 原始消息超过几条才触发摘要

model = init_chat_model("openai:gpt-4o-mini", temperature=0)

SUMMARY_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "把已有摘要和新增对话合并成一段不超过 120 字的中文摘要，"
               "保留姓名、编号、金额、明确偏好；不要编造。"),
    ("human", "已有摘要：\n{summary}\n\n新增对话：\n{new_lines}\n\n新摘要："),
])

CHAT_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "你是一个中文客服助手。较早对话的摘要：\n{summary}"),
    MessagesPlaceholder(variable_name="recent"),
    ("human", "{question}"),
])

summary_chain = SUMMARY_PROMPT | model
chat_chain = CHAT_PROMPT | model


class SummaryBufferMemory:
    def __init__(self, keep_rounds=KEEP_ROUNDS, trigger=TRIGGER):
        self.summary = "（暂无）"
        self.history = InMemoryChatMessageHistory()
        self.keep = keep_rounds * 2
        self.trigger = trigger

    @staticmethod
    def _render(msgs):
        role = {"human": "用户", "ai": "助手"}
        return "\n".join("%s：%s" % (role.get(m.type, m.type), m.content) for m in msgs)

    def recent(self):
        """读：只取最近 keep 条原始消息。"""
        return self.history.messages[-self.keep:]

    def save(self, question, answer):
        """写：追加本轮，必要时把溢出的旧消息压进摘要。"""
        self.history.add_messages([HumanMessage(content=question),
                                   AIMessage(content=answer)])
        msgs = self.history.messages
        if len(msgs) <= self.trigger:
            return

        overflow = msgs[:-self.keep]          # 这一段即将被摘要吸收
        self.summary = summary_chain.invoke({
            "summary": self.summary,
            "new_lines": self._render(overflow),
        }).content

        # 吸收完就从容器里移除，避免下次重复摘要同一段
        kept = msgs[-self.keep:]
        self.history.clear()
        self.history.add_messages(kept)


def ask(mem, question):
    reply = chat_chain.invoke({"summary": mem.summary,
                               "recent": mem.recent(),
                               "question": question})
    mem.save(question, reply.content)
    return reply.content


if __name__ == "__main__":
    mem = SummaryBufferMemory()
    for q in ["你好，我想查订单 12345 的状态",
              "这个订单是上周五下的",
              "我现在急着用，能加急吗",
              "等等，我记错了，应该是 12346",
              "你们的退货政策是怎样的",
              "我最后确认的订单号是多少？"]:
        print("用户：", q)
        print("助手：", ask(mem, q), "\n")

    print("摘要：", mem.summary)
    print("保留原始消息 %d 条" % len(mem.history.messages))

# -*- coding: utf-8 -*-
"""
trim_messages：按 token 预算修剪历史
======================================
手写切片只能按「条数」切，而模型的限制是按 token 算的 ——
一条 500 字的长消息和一条「嗯」占的 token 差了两个量级。
trim_messages 直接按 token 预算裁，并且能保住人设和消息配对。

几个关键参数：
  max_tokens     预算上限
  strategy       "last" 保留最后若干条（多轮对话用它）；"first" 保留开头
  token_counter  怎么数 token；传模型对象就用该模型的分词口径
  include_system 保留开头的 SystemMessage，人设不会被裁掉
  start_on       裁完后第一条从什么角色开始，"human" 可避免以 AI 回复开头
  allow_partial  允许把一条消息截一半，默认 False
"""

from langchain.chat_models import init_chat_model
from langchain_core.messages import (AIMessage, HumanMessage, SystemMessage,
                                     trim_messages)
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough

model = init_chat_model("openai:gpt-4o-mini", temperature=0)

MESSAGES = [
    SystemMessage(content="你是一个中文助手，回答简洁。"),
    HumanMessage(content="你好，我叫孙小空"),
    AIMessage(content="你好孙小空，很高兴认识你。"),
    HumanMessage(content="我有两个师弟，猪小戒和沙小僧"),
    AIMessage(content="听起来是个热闹的团队。"),
    HumanMessage(content="我今年考上了一本"),
    AIMessage(content="恭喜你，这是很了不起的成绩。"),
    HumanMessage(content="我叫什么名字？"),
]


def trim_by_tokens(budget=80):
    """按 token 预算裁，人设保留，从用户消息开始。"""
    return trim_messages(
        MESSAGES,
        max_tokens=budget,
        strategy="last",
        token_counter=model,
        include_system=True,
        start_on="human",
        allow_partial=False,
    )


def trim_by_count(n=4):
    """临时想按条数裁：token_counter 传 len，一条就算一个单位。"""
    return trim_messages(MESSAGES, max_tokens=n, strategy="last",
                         token_counter=len, include_system=True,
                         start_on="human")


# 修剪是链上的一个普通环节，可以直接串进 LCEL
prompt = ChatPromptTemplate.from_messages([
    MessagesPlaceholder(variable_name="history"),
    ("human", "{question}"),
])
trimmer = trim_messages(max_tokens=200, strategy="last", token_counter=model,
                        include_system=True, start_on="human")
chain = RunnablePassthrough.assign(history=lambda x: trimmer.invoke(x["history"])) \
    | prompt | model

if __name__ == "__main__":
    for m in trim_by_tokens(80):
        print("[%s] %s" % (m.type, m.content))
    print("---")
    print("按条数裁剩 %d 条" % len(trim_by_count(4)))

    # 裁到只剩人设 + 最后一问时，模型就答不出名字了 —— 预算给小了
    print(chain.invoke({"history": MESSAGES[:-1],
                        "question": "我叫什么名字？"}).content)

# -*- coding: utf-8 -*-
"""
Token 预算：一次请求的额度到底怎么分
=======================================
上下文窗口是一个硬上限，它同时装下这些东西：

    人设 + 历史 + 本轮提问 + 工具描述  ≤  窗口 - 预留给回复的额度

所以「历史能带多少」不是拍脑袋定的，而是减出来的。
这个文件把减法写成代码：先算固定开销，剩下的才是历史的预算。
"""

from langchain.chat_models import init_chat_model
from langchain_core.messages import (AIMessage, HumanMessage, SystemMessage,
                                     trim_messages)

CONTEXT_WINDOW = 8192      # 模型的上下文窗口，按你实际用的模型填
RESERVE_FOR_REPLY = 1024   # 给回复留出的额度，不留就可能被截断
SAFETY = 256               # 缓冲：不同 provider 计费口径略有差异

model = init_chat_model("openai:gpt-4o-mini", temperature=0)

SYSTEM = SystemMessage(content="你是一个中文客服助手，回答控制在五句话以内。")


def count(messages):
    """用模型自己的口径数 token，比按字数估准得多。"""
    return model.get_num_tokens_from_messages(messages)


def history_budget(question: str) -> int:
    """算出这一轮历史最多能占多少 token。"""
    fixed = count([SYSTEM, HumanMessage(content=question)])
    budget = CONTEXT_WINDOW - RESERVE_FOR_REPLY - SAFETY - fixed
    return max(budget, 0)


def build_request(history, question):
    """按预算裁历史，拼出这一轮真正发出去的消息列表。"""
    budget = history_budget(question)
    kept = trim_messages(
        [SYSTEM] + list(history),
        max_tokens=budget,
        strategy="last",
        token_counter=model,
        include_system=True,
        start_on="human",
    )
    payload = kept + [HumanMessage(content=question)]

    used = count(payload)
    if used > CONTEXT_WINDOW - RESERVE_FOR_REPLY:
        # 裁完还超，说明单条消息本身就过长：该走摘要或分段，而不是继续裁
        raise ValueError("单轮输入 %d token 仍超预算，请改用摘要策略" % used)

    return payload, {"budget": budget, "used": used,
                     "dropped": len(history) + 1 - len(kept)}


if __name__ == "__main__":
    history = []
    for i in range(1, 21):
        history.append(HumanMessage(content="第 %d 个问题，内容比较长" % i * 3))
        history.append(AIMessage(content="第 %d 个回答，同样写得比较详细" % i * 3))

    payload, stat = build_request(history, "把我前面问过的第一个问题复述一遍")
    print("历史预算 %(budget)d token，本次实际使用 %(used)d token，丢弃 %(dropped)d 条" % stat)
    print("发出去的消息共 %d 条" % len(payload))

    # 超长对话的三种处理，按代价从低到高：
    #   1. 裁：trim_messages 按预算留最近若干条，最便宜，丢远期信息
    #   2. 摘：把溢出的一段压成摘要，多一次模型调用，保住关键事实
    #   3. 存：把全文写进外部存储，需要时再按问题检索回来，最贵也最完整

# -*- coding: utf-8 -*-
"""
多用户多会话隔离：session_id 的工厂函数怎么写
================================================
线上不是只有一个人在聊天。工厂函数是隔离的唯一入口，写歪了就会串会话。

三条规则：
  1. key 必须能唯一定位到「谁的哪一段对话」，通常是 用户ID + 会话ID
  2. 工厂函数只负责「给 key 返回容器」，不要在里面做裁剪、摘要、鉴权
  3. 进程内字典只适合本机单进程；多副本部署必须换成外部存储
"""

from langchain.chat_models import init_chat_model
from langchain_core.chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.runnables.utils import ConfigurableFieldSpec

model = init_chat_model("openai:gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个电商客服助手，用中文友好作答。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{question}"),
])

_store: dict[tuple[str, str], BaseChatMessageHistory] = {}


def get_history(user_id: str, conversation_id: str) -> BaseChatMessageHistory:
    """两个维度共同决定一块备忘板：同一个人的不同会话也要互相隔离。"""
    key = (user_id, conversation_id)
    if key not in _store:
        _store[key] = InMemoryChatMessageHistory()
    return _store[key]


# 默认只认 session_id 一个字段。想用多个字段，就把它们声明出来
conversation = RunnableWithMessageHistory(
    prompt | model,
    get_history,
    input_messages_key="question",
    history_messages_key="history",
    history_factory_config=[
        ConfigurableFieldSpec(
            id="user_id", annotation=str, name="用户 ID",
            description="登录后的稳定用户标识", default="", is_shared=True,
        ),
        ConfigurableFieldSpec(
            id="conversation_id", annotation=str, name="会话 ID",
            description="同一用户的第几段对话", default="", is_shared=True,
        ),
    ],
)


def ask(user_id, conversation_id, question):
    cfg = {"configurable": {"user_id": user_id, "conversation_id": conversation_id}}
    return conversation.invoke({"question": question}, config=cfg).content


if __name__ == "__main__":
    print(ask("u1001", "c1", "我的订单号是 12345"))
    print(ask("u1001", "c1", "我刚说的订单号是多少？"))     # 记得 12345

    # 同一个人换一段会话：查不到上一段的订单号
    print(ask("u1001", "c2", "我刚说的订单号是多少？"))

    # 另一个人：更不可能看到别人的订单号
    print(ask("u2002", "c1", "我刚说的订单号是多少？"))

    print("当前共 %d 块独立历史" % len(_store))            # 3

# -*- coding: utf-8 -*-
"""
生产环境的持久化历史：自己实现一个 BaseChatMessageHistory
============================================================
进程内字典有三个致命问题：重启即丢、多副本各存各的、内存只涨不降。
换成外部存储只需要实现三个成员，链上的其它代码一个字都不用改。

要实现的接口：
    messages          属性，返回 List[BaseMessage]
    add_messages(x)   追加若干条
    clear()           清空本会话

社区也提供了现成的 RedisChatMessageHistory，思路与这份实现一致；
自己写一份的好处是能控制 key 规则、过期时间和序列化格式。
"""

import json
import os

import redis
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import (BaseMessage, messages_from_dict,
                                     messages_to_dict)

# 连接信息一律走环境变量，口令不进源码、不进版本库
_client = redis.Redis.from_url(
    os.environ.get("REDIS_URL", "redis://127.0.0.1:6379/0"),
    decode_responses=True,
)

TTL_SECONDS = 60 * 60 * 24 * 7        # 七天没人说话就自动回收


class RedisMessageHistory(BaseChatMessageHistory):
    """用一个 Redis List 存一段会话，每个元素是一条序列化后的消息。"""

    def __init__(self, session_id: str, client=None, ttl: int = TTL_SECONDS):
        self.session_id = session_id
        self.client = client or _client
        self.ttl = ttl

    @property
    def key(self) -> str:
        # 加业务前缀，避免和别的数据撞键
        return "chat:history:%s" % self.session_id

    @property
    def messages(self) -> list[BaseMessage]:
        raw = self.client.lrange(self.key, 0, -1)
        return messages_from_dict([json.loads(item) for item in raw])

    def add_messages(self, messages: list[BaseMessage]) -> None:
        if not messages:
            return
        payload = [json.dumps(d, ensure_ascii=False)
                   for d in messages_to_dict(messages)]
        pipe = self.client.pipeline()
        pipe.rpush(self.key, *payload)
        pipe.expire(self.key, self.ttl)      # 每次写入都续期
        pipe.execute()

    def clear(self) -> None:
        self.client.delete(self.key)

    def tail(self, n: int) -> list[BaseMessage]:
        """只取最近 n 条：窗口策略在存储层就能完成，不用把全量读回内存。"""
        raw = self.client.lrange(self.key, -n, -1)
        return messages_from_dict([json.loads(item) for item in raw])


def get_history(session_id: str) -> BaseChatMessageHistory:
    """把它交给 RunnableWithMessageHistory，唯一的改动就是这一个函数。"""
    return RedisMessageHistory(session_id)


if __name__ == "__main__":
    h = get_history("u1001-c1")
    h.clear()
    h.add_messages([])                       # 空列表要安全返回
    from langchain_core.messages import AIMessage, HumanMessage

    h.add_messages([HumanMessage(content="我的订单号是 12346"),
                    AIMessage(content="已记录订单号 12346。")])

    print("读回 %d 条" % len(h.messages))
    print("最近 1 条：", h.tail(1)[0].content)
    print("剩余存活时间：%s 秒" % _client.ttl(h.key))

# -*- coding: utf-8 -*-
"""
完整案例：带记忆的电商客服机器人
==================================
把前面拆开讲的零件装成一个能跑的东西：

  · 混合记忆：摘要 + 最近 K 轮，长对话也不会顶穿窗口
  · 多用户隔离：user_id + conversation_id 两级 key
  · 可换存储：把 HISTORY_FACTORY 换成 Redis 版本即可上生产
  · 事实卡片：订单号这类关键信息单独记一份，不让摘要把它抹掉

运行后模拟一段六轮对话，最后一问考察它记不记得改过的订单号。
"""

import os

from langchain.chat_models import init_chat_model
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

KEEP = 4                 # 最近保留几条原始消息
TRIGGER = 8              # 超过几条才触发摘要

model = init_chat_model(
    os.environ.get("CHAT_MODEL", "openai:gpt-4o-mini"), temperature=0.3)

CHAT_PROMPT = ChatPromptTemplate.from_messages([
    ("system",
     "你是电商平台的客服助手，用中文友好、专业地回答。\n"
     "已确认的事实：{facts}\n"
     "较早对话的摘要：{summary}"),
    MessagesPlaceholder(variable_name="recent"),
    ("human", "{question}"),
])

SUMMARY_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "把已有摘要与新增对话合并成不超过 120 字的中文摘要，"
               "保留订单号、金额、时间与承诺；不要编造。"),
    ("human", "已有摘要：\n{summary}\n\n新增对话：\n{lines}\n\n新摘要："),
])

chat_chain = CHAT_PROMPT | model | StrOutputParser()
summary_chain = SUMMARY_PROMPT | model | StrOutputParser()


def HISTORY_FACTORY(key):
    """换成 redis_history.get_history 就是生产版本，其余代码不动。"""
    return _mem.setdefault(key, InMemoryChatMessageHistory())


_mem = {}
_summary = {}
_facts = {}


def _render(msgs):
    role = {"human": "用户", "ai": "客服"}
    return "\n".join("%s：%s" % (role.get(m.type, m.type), m.content) for m in msgs)


def remember_fact(key, name, value):
    """关键事实单独存一份，它不参与裁剪也不进摘要，因此永远不会丢。"""
    _facts.setdefault(key, {})[name] = value


def ask(user_id, conversation_id, question):
    key = (user_id, conversation_id)
    history = HISTORY_FACTORY(key)

    answer = chat_chain.invoke({
        "facts": _facts.get(key) or "（暂无）",
        "summary": _summary.get(key, "（暂无）"),
        "recent": history.messages[-KEEP:],
        "question": question,
    })

    history.add_messages([HumanMessage(content=question),
                          AIMessage(content=answer)])

    msgs = history.messages
    if len(msgs) > TRIGGER:
        overflow = msgs[:-KEEP]
        _summary[key] = summary_chain.invoke({
            "summary": _summary.get(key, "（暂无）"),
            "lines": _render(overflow),
        })
        kept = msgs[-KEEP:]
        history.clear()
        history.add_messages(kept)

    return answer


if __name__ == "__main__":
    U, C = "u1001", "c1"
    dialogue = [
        "你好，我想查订单 12345 的状态",
        "这个订单是上周五下的",
        "我现在急着用，能加急处理吗",
        "等等，我记错了，应该是 12346",
        "你们的退货政策是怎样的",
        "我最终确认的订单号是多少？",
    ]

    for i, q in enumerate(dialogue, 1):
        if "12346" in q:
            remember_fact((U, C), "订单号", "12346")
        print("第 %d 轮 用户：%s" % (i, q))
        print("        客服：%s\n" % ask(U, C, q))

    print("摘要：", _summary.get((U, C), "（未触发）"))
    print("事实卡片：", _facts.get((U, C)))
    print("保留原始消息 %d 条" % len(HISTORY_FACTORY((U, C)).messages))

# -*- coding: utf-8 -*-
"""
带记忆的对话骨架 —— 复制后只改 TODO 处
==========================================
四种策略写在同一份骨架里，改一个常量就能切换：

    STRATEGY = "buffer"   全量保存
             = "window"   只留最近 K 轮
             = "trim"     按 token 预算修剪
             = "summary"  摘要 + 最近 K 轮

存储实现、模型、人设各自只有一个改动点，其余代码不用动。
"""

import os

from langchain.chat_models import init_chat_model
from langchain_core.chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory
from langchain_core.messages import AIMessage, HumanMessage, trim_messages
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# TODO(1)：选策略与参数
STRATEGY = "window"
KEEP_MESSAGES = 6          # window / summary 保留的原始消息条数
TOKEN_BUDGET = 1500        # trim 策略的 token 预算
SUMMARY_TRIGGER = 12       # summary 策略：超过多少条才压缩

# TODO(2)：换模型。需要密钥的 provider 一律读环境变量
model = init_chat_model(
    os.environ.get("CHAT_MODEL", "openai:gpt-4o-mini"), temperature=0)

# TODO(3)：换人设。历史插槽的名字要和下面 recent 保持一致
PROMPT = ChatPromptTemplate.from_messages([
    ("system", "你是一个中文助手。较早对话的摘要：{summary}"),
    MessagesPlaceholder(variable_name="recent"),
    ("human", "{question}"),
])

SUMMARY_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "把已有摘要与新增对话合并成不超过 120 字的中文摘要，保留关键事实，不要编造。"),
    ("human", "已有摘要：\n{summary}\n\n新增对话：\n{lines}\n\n新摘要："),
])

chat_chain = PROMPT | model | StrOutputParser()
summary_chain = SUMMARY_PROMPT | model | StrOutputParser()

_store: dict[str, BaseChatMessageHistory] = {}
_summary: dict[str, str] = {}


# TODO(4)：换存储。上生产时改成 Redis / 数据库实现即可，其余代码不动
def get_history(session_key: str) -> BaseChatMessageHistory:
    return _store.setdefault(session_key, InMemoryChatMessageHistory())


def _render(msgs):
    role = {"human": "用户", "ai": "助手"}
    return "\n".join("%s：%s" % (role.get(m.type, m.type), m.content) for m in msgs)


def read_recent(session_key):
    """读：按策略决定这一轮把哪些历史发回去。"""
    msgs = get_history(session_key).messages
    if STRATEGY == "buffer":
        return msgs
    if STRATEGY in ("window", "summary"):
        return msgs[-KEEP_MESSAGES:]
    if STRATEGY == "trim":
        return trim_messages(msgs, max_tokens=TOKEN_BUDGET, strategy="last",
                             token_counter=model, include_system=True,
                             start_on="human")
    raise ValueError("未知策略：%s" % STRATEGY)


def write_back(session_key, question, answer):
    """写：把这一轮追加进去；summary 策略下顺便压缩溢出的部分。"""
    history = get_history(session_key)
    history.add_messages([HumanMessage(content=question),
                          AIMessage(content=answer)])

    if STRATEGY != "summary":
        return
    msgs = history.messages
    if len(msgs) <= SUMMARY_TRIGGER:
        return

    overflow = msgs[:-KEEP_MESSAGES]
    _summary[session_key] = summary_chain.invoke({
        "summary": _summary.get(session_key, "（暂无）"),
        "lines": _render(overflow),
    })
    kept = msgs[-KEEP_MESSAGES:]
    history.clear()
    history.add_messages(kept)


def ask(session_key: str, question: str) -> str:
    answer = chat_chain.invoke({
        "summary": _summary.get(session_key, "（暂无）"),
        "recent": read_recent(session_key),
        "question": question,
    })
    write_back(session_key, question, answer)
    return answer


if __name__ == "__main__":
    # TODO(5)：换成你的会话标识，通常是 用户ID:会话ID
    key = "u1001:c1"
    print(ask(key, "你好，我叫孙小空"))
    print(ask(key, "我叫什么名字？"))
    print("历史 %d 条" % len(get_history(key).messages))

# -*- coding: utf-8 -*-
"""
生产环境的持久化历史：自己实现一个 BaseChatMessageHistory
============================================================
进程内字典有三个致命问题：重启即丢、多副本各存各的、内存只涨不降。
换成外部存储只需要实现三个成员，链上的其它代码一个字都不用改。

要实现的接口：
    messages          属性，返回 List[BaseMessage]
    add_messages(x)   追加若干条
    clear()           清空本会话

社区也提供了现成的 RedisChatMessageHistory，思路与这份实现一致；
自己写一份的好处是能控制 key 规则、过期时间和序列化格式。
"""

import json
import os

import redis
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import (BaseMessage, messages_from_dict,
                                     messages_to_dict)

# 连接信息一律走环境变量，口令不进源码、不进版本库
_client = redis.Redis.from_url(
    os.environ.get("REDIS_URL", "redis://127.0.0.1:6379/0"),
    decode_responses=True,
)

TTL_SECONDS = 60 * 60 * 24 * 7        # 七天没人说话就自动回收


class RedisMessageHistory(BaseChatMessageHistory):
    """用一个 Redis List 存一段会话，每个元素是一条序列化后的消息。"""

    def __init__(self, session_id: str, client=None, ttl: int = TTL_SECONDS):
        self.session_id = session_id
        self.client = client or _client
        self.ttl = ttl

    @property
    def key(self) -> str:
        # 加业务前缀，避免和别的数据撞键
        return "chat:history:%s" % self.session_id

    @property
    def messages(self) -> list[BaseMessage]:
        raw = self.client.lrange(self.key, 0, -1)
        return messages_from_dict([json.loads(item) for item in raw])

    def add_messages(self, messages: list[BaseMessage]) -> None:
        if not messages:
            return
        payload = [json.dumps(d, ensure_ascii=False)
                   for d in messages_to_dict(messages)]
        pipe = self.client.pipeline()
        pipe.rpush(self.key, *payload)
        pipe.expire(self.key, self.ttl)      # 每次写入都续期
        pipe.execute()

    def clear(self) -> None:
        self.client.delete(self.key)

    def tail(self, n: int) -> list[BaseMessage]:
        """只取最近 n 条：窗口策略在存储层就能完成，不用把全量读回内存。"""
        raw = self.client.lrange(self.key, -n, -1)
        return messages_from_dict([json.loads(item) for item in raw])


def get_history(session_id: str) -> BaseChatMessageHistory:
    """把它交给 RunnableWithMessageHistory，唯一的改动就是这一个函数。"""
    return RedisMessageHistory(session_id)


if __name__ == "__main__":
    h = get_history("u1001-c1")
    h.clear()
    h.add_messages([])                       # 空列表要安全返回
    from langchain_core.messages import AIMessage, HumanMessage

    h.add_messages([HumanMessage(content="我的订单号是 12346"),
                    AIMessage(content="已记录订单号 12346。")])

    print("读回 %d 条" % len(h.messages))
    print("最近 1 条：", h.tail(1)[0].content)
    print("剩余存活时间：%s 秒" % _client.ttl(h.key))