8. 对话状态与记忆

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
 
llm = ChatOpenAI(model="gpt-4o-mini")
 
# 第一个问题
response1 = llm.invoke([HumanMessage(content="我的名字是 Alice")])
print(response1.content)  # 输出: 很高兴认识你,Alice!
 
# 第二个问题(没有发送历史)
response2 = llm.invoke([HumanMessage(content="我的名字是什么?")])
print(response2.content)  # 输出: 我不知道你的名字...

# 用户询问一个主题
response1 = llm.invoke([HumanMessage(content="告诉我关于 Python 的信息")])
print(response1.content)  
# 输出: Python 是一种以可读性著称的高级编程语言...
 
# 用户用代词跟进
response2 = llm.invoke([HumanMessage(content="它的主要特性是什么?")])
print(response2.content)  
# 输出: 我很乐意帮忙!你能具体说明你在问什么吗?

用户: "我的订单 #12345 有问题"
代理: "很抱歉听到这个消息。有什么问题?"
用户: "配送地址错了"
代理: "我可以帮你解决。你能提供你的订单号吗?"
用户: "我刚告诉你了..."

from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
 
# SystemMessage: 模型行为的指令
system_msg = SystemMessage(content="你是一个专门研究 Python 编程的有帮助的助手。")
 
# HumanMessage: 用户输入
user_msg = HumanMessage(content="如何在 Python 中读取文件?")
 
# AIMessage: 模型的响应
# (实际上,LangChain 会将模型的响应包装在这个对象中 - 这里仅用于说明)
ai_msg = AIMessage(content="你可以使用带有上下文管理器的 `open()` 函数...")

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
 
llm = ChatOpenAI(model="gpt-4o-mini")
 
# 构建对话历史
# SystemMessage 设置指令(在开始时一次)
# 然后: 用户输入 → AI 响应 → 用户输入(自然对话流程)
messages = [
    SystemMessage(content="你是一个简洁的 Python 导师。"),
    HumanMessage(content="什么是列表推导式?"),
    AIMessage(content="列表推导式是创建列表的简洁方式: [x*2 for x in range(5)]"),
    HumanMessage(content="你能给我展示一个更复杂的例子吗?")
]
 
# 发送完整历史与新问题
response = llm.invoke(messages)
print(response.content)

当然!这是一个过滤和转换的列表推导式:
[x**2 for x in range(10) if x % 2 == 0]
这会产生:
[0, 4, 16, 36, 64]

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
 
# 创建消息历史存储
history = InMemoryChatMessageHistory()
 
# 一次添加多条消息
history.add_messages([
    SystemMessage(content="你是一个有帮助的 Python 导师。"),
    HumanMessage(content="Python 中的装饰器是什么?")
])
 
# 逐条添加消息
history.add_message(AIMessage(content="装饰器是修改另一个函数行为的函数..."))
history.add_message(HumanMessage(content="你能展示一个例子吗?"))
 
# 检索所有消息
messages = history.messages

from langchain_openai import ChatOpenAI
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import SystemMessage, HumanMessage
 
llm = ChatOpenAI(model="gpt-4o-mini")
history = InMemoryChatMessageHistory()
 
# 设置系统消息并添加到历史(在开始时执行一次)
system_msg = SystemMessage(content="你是一个有帮助的 Python 导师。")
history.add_message(system_msg)
 
# 聊天函数
def chat(user_input):
    """处理用户输入,发送到 LLM,并自动管理对话历史"""
    # 将用户输入添加到历史
    history.add_message(HumanMessage(content=user_input))
    
    # 发送用户输入以及之前的对话历史
    response = llm.invoke(history.messages)
    
    # 将模型响应添加到历史
    history.add_message(response)
    
    return response.content
 
# 模拟对话
print(chat("什么是 lambda 函数?"))
print(chat("给我展示一个例子"))  # 模型记住上下文
print(chat("它与常规函数有什么区别?"))  # 仍然记得

lambda 函数是用 lambda 关键字定义的匿名函数...
 
这是一个例子: square = lambda x: x**2
你可以这样使用它: square(5) # 返回 25
 
lambda 函数限于单个表达式,而常规函数...

# chapter3_cli.py (原始 - 无状态)
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
 
llm = ChatOpenAI(model="gpt-4o-mini")
 
def chat_loop():
    print("聊天已开始。输入 'quit' 退出。\n")
    while True:
        user_input = input("你: ")
        if user_input.lower() == "quit":
            break
        
        # 无状态 - 没有历史
        response = llm.stream([HumanMessage(content=user_input)])
        print("AI: ", end="", flush=True)
        for chunk in response:
            print(chunk.content, end="", flush=True)
        print("\n")
 
if __name__ == "__main__":
    chat_loop()

# chapter8_cli.py (带记忆)
from langchain_openai import ChatOpenAI
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
 
llm = ChatOpenAI(model="gpt-4o-mini")
 
def chat_loop():
    print("聊天已开始。输入 'quit' 退出。\n")
    
    # 创建历史并添加系统消息
    history = InMemoryChatMessageHistory()
    history.add_message(SystemMessage(content="你是一个有帮助的助手。"))
    
    while True:
        user_input = input("你: ")
        if user_input.lower() == "quit":
            break
        
        # 将用户消息添加到历史
        history.add_message(HumanMessage(content=user_input))
        
        # 流式响应
        print("AI: ", end="", flush=True)
        full_response = ""
        for chunk in llm.stream(history.messages):
            print(chunk.content, end="", flush=True)
            full_response += chunk.content
        print("\n")
        
        # 将 AI 响应添加到历史
        history.add_message(AIMessage(content=full_response))
 
if __name__ == "__main__":
    chat_loop()

聊天已开始。输入 'quit' 退出。
 
你: 我的名字是 Alice
AI: 很高兴认识你,Alice!今天我能帮你什么?
 
你: 我的名字是什么?
AI: 你的名字是 Alice。
 
你: 我刚才问了你什么?
AI: 你问我你的名字是什么。
 
你: quit

# [开发] 本地内存
# from langchain_core.chat_history import InMemoryChatMessageHistory
# history = InMemoryChatMessageHistory()
 
# [生产] PostgreSQL 持久化存储
import psycopg
from langchain_postgres import PostgresChatMessageHistory
 
# 创建 PostgreSQL 连接
sync_connection = psycopg.connect(
    "postgresql://user:password@10.1.1.100:5432/agent_db",
    autocommit=True,
)
 
# 指定数据库连接和会话 ID
# session_id: 用于识别对话的唯一键
#   - 按用户管理: session_id = user_id (每个用户一个对话)
#   - 按会话管理: session_id = uuid (每个对话新 ID)
history = PostgresChatMessageHistory(
    table_name="chat_history",
    session_id="user_123",
    sync_connection=sync_connection
)
 
# --- 无论存储类型如何,接口都相同 ---
history.add_message(HumanMessage(content="显示我之前的订单。"))
print(history.messages)

import tiktoken
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langchain_core.messages.utils import trim_messages
 
# 获取你的模型的分词器(不同模型系列使用不同的分词器)
# 提供模型名称以获取适当的分词器
# Claude 模型: 使用 Anthropic 的分词器(tiktoken 是 OpenAI 特定的)
enc = tiktoken.encoding_for_model("gpt-4o")
 
def token_counter(msg: BaseMessage) -> int:
    """计算单条消息中的令牌数。"""
    return len(enc.encode(msg.content or ""))
 
# 创建对话历史
messages = [
    SystemMessage(content="你是一个有帮助的助手。"),
    HumanMessage(content="嗨!"),
    AIMessage(content="你好!我能帮你什么?"),
    HumanMessage(content="2+2 是多少?"),
    AIMessage(content="2+2 等于 4。"),
    HumanMessage(content="3+3 是多少?"),
    AIMessage(content="3+3 等于 6。"),
    HumanMessage(content="4+4 是多少?"),
]
 
# 仅保留最大令牌计数内的消息
trimmed = trim_messages(
    messages,
    max_tokens=30,
    token_counter=token_counter,
    include_system=True,
)
 
print(f"原始: {len(messages)} 条消息")
print(f"修剪后: {len(trimmed)} 条消息")
for m in trimmed:
    print(f"{type(m).__name__}: {m.content}")

import tiktoken
from langchain_openai import ChatOpenAI
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import SystemMessage, HumanMessage, BaseMessage
from langchain_core.messages.utils import trim_messages
 
llm = ChatOpenAI(model="gpt-4o-mini")
history = InMemoryChatMessageHistory()
enc = tiktoken.encoding_for_model("gpt-4o-mini")
 
def token_counter(msg: BaseMessage) -> int:
    """计算单条消息中的令牌数。"""
    return len(enc.encode(msg.content or ""))
 
def chat_with_trimming(user_input: str, max_tokens: int = 1000) -> str:
    """带自动历史修剪的聊天。"""
    history.add_message(HumanMessage(content=user_input))
    
    system_msg = SystemMessage(content="你是一个有帮助的助手。")
    all_messages = [system_msg] + history.messages
    
    # 修剪到最大令牌计数
    trimmed_messages = trim_messages(
        all_messages,
        max_tokens=max_tokens,
        token_counter=token_counter,
        include_system=True,
    )
    
    response = llm.invoke(trimmed_messages)
    history.add_message(response)
    
    return response.content
 
# 使用示例
print(chat_with_trimming("我正在计划去日本旅行"))
print(chat_with_trimming("我应该在东京参观什么?"))
print(chat_with_trimming("我应该在那里待几天?"))
# 即使历史增长,也只有最大令牌计数内的最近对话被发送到 LLM