Easy-lang的学习笔记

本章主要从状态管理、外部行动维度来学习实现agent的带记忆、能交互外部系统的能力。

3.1 Memory(状态管理层):让大模型具有记忆功能

虽然每一次的llm调用时memory都是无状态，独立的，但我们可以通过结构化的方式存储、管理对话历史，让AI具备“记忆能力”。

主要有两个核心的动作：

存储(Save)：将每一轮的用户输入（HumanMessage）和AI输出（AIMessage）保存到指定存储介质（内存、数据库等）。
提取(Load)：新一轮对话时，从存储介质中提取历史对话，注入到Prompt中供LLM参考。

这里主要学习的是LCEL架构，主要实现为三种记忆形式：

全量记忆：完整保存所有对话历史，适用于短对话场景
窗口记忆：仅保留最近N轮对话，控制Token消耗
摘要记忆：通过LLM生成对话摘要替代完整历史，平衡上下文连贯性与效率

全量记忆

全量记忆适合短对话

# 1. 定义提示词模板（包含历史消息占位符）
full_memory_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是友好的对话助手，需基于完整的历史对话回答用户问题。"),
    MessagesPlaceholder(variable_name="chat_history"),  # 历史消息占位符
    ("human", "{user_input}")  # 用户当前输入
])

# 2. 构建基础链（提示词 + LLM）
base_chain = full_memory_prompt | llm

# 3. 会话历史存储（内存模式，生产环境可替换为数据库存储）
full_memory_store = {}

# 4. 定义会话历史获取函数（核心：返回完整历史）
def get_full_memory_history(session_id: str) -> BaseChatMessageHistory:
    """根据session_id获取会话历史，不存在则创建新的历史记录"""
    if session_id not in full_memory_store:
        full_memory_store[session_id] = InMemoryChatMessageHistory()
    return full_memory_store[session_id]

# 5. 构建带全量记忆的对话链
full_memory_chain = RunnableWithMessageHistory(
    runnable=base_chain,
    get_session_history=get_full_memory_history,
    input_messages_key="user_input",  # 输入中用户问题的键名
    history_messages_key="chat_history"  # 传入提示词的历史消息键名
)

# 测试多轮对话（指定session_id=user_001，隔离不同用户）
config = {"configurable": {"session_id": "user_001"}}

# 第一轮对话
response1 = full_memory_chain.invoke({"user_input": "我叫小明，喜欢编程"}, config=config)
print("助手回复1：", response1.content)
# 输出示例：你好小明！编程是一项很有创造力的技能，你平时常用什么编程语言呢？

# 第二轮对话（验证记忆：询问历史信息）
response2 = full_memory_chain.invoke({"user_input": "我刚才说我喜欢什么？"}, config=config)
print("助手回复2：", response2.content)
# 输出示例：你刚才说你喜欢编程呀～

# 查看完整历史记录
print("\n全量记忆的对话历史：")
for msg in get_full_memory_history("user_001").messages:
    print(f"{msg.type}: {msg.content}")

MessagesPlaceholder(variable_name="chat_history") ：一个历史消息占位符。这是实现对话记忆的关键。它不在模板中写死任何内容，而是在程序运行时，动态地将之前的对话记录（比如用户之前问了什么，AI回答了什么）插入到这个位置。这样，AI在回答新问题时就能参考上下文，实现连贯的多轮对话

base_chain = full_memory_prompt | llm ：使用管道操作符 |将两个组件连接起来，形成了一个简单的处理链，其含义是将前一个组件的输出，作为后一个组件的输入
其实就是拆分成三个流程：

1. 执行prompt：
2.管道执行：这个生成好的消息列表被自动传递给 llm
1. 执行llm：根据收到的消息列表，生成一段连贯且符合上下文的回答。

执行结果：

助手回复1： 你好，小明！很高兴认识你。喜欢编程真棒，你平时更喜欢哪种编程语言，或者在做哪一类项目呢？
助手回复2： 你刚才说你喜欢编程。

全量记忆的对话历史：
human: 我叫小明，喜欢编程
ai: 你好，小明！很高兴认识你。喜欢编程真棒，你平时更喜欢哪种编程语言，或者在做哪一类项目呢？
human: 我刚才说我喜欢什么？
ai: 你刚才说你喜欢编程。

窗口记忆

它只保留最近的N轮对话（N用k参数控制），早期的对话会自动丢弃，这样能有效控制文字量，适合客服、长期陪伴等长对话场景。

# 1. 定义提示词模板（与全量记忆通用，可复用）
window_memory_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是友好的对话助手，需基于最近的对话历史回答用户问题。"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{user_input}")
])

# 2. 构建基础链
window_base_chain = window_memory_prompt | llm

# 3. 会话历史存储
window_memory_store = {}
WINDOW_SIZE = 2  # 保留最近2轮对话（即最近4条消息：用户-助手-用户-助手）

# 4. 定义带窗口限制的会话历史获取函数
def get_window_memory_history(session_id: str) -> BaseChatMessageHistory:
    """获取会话历史，仅保留最近WINDOW_SIZE轮对话"""
    if session_id not in window_memory_store:
        window_memory_store[session_id] = InMemoryChatMessageHistory()
    
    # 获取完整历史，截取最近WINDOW_SIZE轮（每轮2条消息）
    history = window_memory_store[session_id]
    if len(history.messages) > 2 * WINDOW_SIZE:
        # 截取后WINDOW_SIZE轮消息（保留最新的）
        history.messages = history.messages[-2 * WINDOW_SIZE:]
    return history

# 5. 构建带窗口记忆的对话链
window_memory_chain = RunnableWithMessageHistory(
    runnable=window_base_chain,
    get_session_history=get_window_memory_history,
    input_messages_key="user_input",
    history_messages_key="chat_history"
)
# 测试多轮对话（session_id=user_002，与全量记忆会话隔离）
config = {"configurable": {"session_id": "user_002"}}

# 模拟5轮对话，验证窗口记忆的截断效果
inputs = [
    "我叫小红",
    "我喜欢画画",
    "我来自上海",
    "我是一名学生",
    "我刚才说我来自哪里？",  # 第5轮：询问第3轮的信息，验证窗口截断
    "我叫什么名字？"  # 第6轮：询问第1轮的信息，验证窗口记忆
]

for i, user_input in enumerate(inputs, 1):
    response = window_memory_chain.invoke({"user_input": user_input}, config=config)
    print(f"\n第{i}轮 - 助手回复：", response.content)

# 查看窗口记忆的最终历史（仅保留最近2轮）
print("\n窗口记忆的最终对话历史（最近2轮）：")
for msg in get_window_memory_history("user_002").messages:
    print(f"{msg.type}: {msg.content}")

执行结果：


我刚才称呼你“小红”是不准确的，如果你愿意，可以告诉我你叫什么，我会记住在这段对话里使用。

窗口记忆的最终对话历史（最近2轮）：
human: 我刚才说我来自哪里？
ai: 你刚才说你来自**上海**。
human: 我叫什么名字？
ai: 你还没有告诉我你的名字。  

我刚才称呼你“小红”是不准确的，如果你愿意，可以告诉我你叫什么，我会记住在这段对话里使用。

摘要记忆

“摘要记忆”：它不保存对话原文，而是用LLM把历史对话总结成一段简洁的摘要。既能保留核心信息，又能最大程度节省文字量，缺点是可能会丢失一些细节（比如具体的数字、名字）。

# 1. 定义摘要生成提示词（用于压缩对话历史）
summary_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是对话摘要助手，需简洁总结以下对话的核心信息（包含用户身份、偏好、关键问题等），不超过50字。"),
    ("human", "对话历史：{chat_history_text}\n请生成摘要：")
])

# 2. 构建摘要生成链（输入完整历史文本，输出摘要）
summary_chain = summary_prompt | llm

# 3. 定义对话记忆提示词（注入摘要而非完整历史）
summary_memory_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是友好的对话助手，需基于对话摘要回答用户问题，摘要包含核心上下文信息。"),
    ("system", "对话摘要：{chat_summary}"),  # 注入摘要
    ("human", "{user_input}")
])

# 4. 构建基础对话链（提示词 + LLM）
summary_base_chain = (
    RunnablePassthrough.assign(
        chat_summary=lambda x: summary_chain.invoke(
            {
                "chat_history_text": "\n".join(
                    [f"{msg.type}: {msg.content}" for msg in x["chat_history"]]
                )
            }
        ).content
    )
    | summary_memory_prompt
    | llm
)

# 5. 会话历史存储（保存完整历史用于生成摘要）
summary_memory_store = {}

# 6. 定义会话历史获取函数
def get_summary_memory_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in summary_memory_store:
        summary_memory_store[session_id] = InMemoryChatMessageHistory()
    return summary_memory_store[session_id]

# 7. 构建带摘要记忆的对话链
summary_memory_chain = RunnableWithMessageHistory(
    runnable=summary_base_chain,
    get_session_history=get_summary_memory_history,
    input_messages_key="user_input",
    history_messages_key="chat_history"  # 传入完整历史用于生成摘要
)

# 测试多轮对话（session_id=user_003）
config = {"configurable": {"session_id": "user_003"}}

# 多轮对话输入
inputs = [
    "我叫小李，是一名产品经理",
    "我负责一款电商APP的迭代",
    "最近在优化用户下单流程",
    "遇到了用户流失率高的问题",
    "你能给我一些优化建议吗？"
]

for i, user_input in enumerate(inputs, 1):
    response = summary_memory_chain.invoke({"user_input": user_input}, config=config)
    print(f"\n第{i}轮 - 助手回复：", response.content)

# 查看完整历史与最终摘要
history = get_summary_memory_history("user_003")
print("\n摘要记忆的完整对话历史：")
for msg in history.messages:
    print(f"{msg.type}: {msg.content}")

# 单独生成最终摘要验证
final_summary = summary_chain.invoke({
    "chat_history_text": "\n".join([f"{msg.type}: {msg.content}" for msg in history.messages])
}).content
print(f"\n最终对话摘要：{final_summary}")
# 输出示例：摘要：小李，产品经理，负责电商APP迭代，优化下单流程时遇用户流失率高问题，寻求建议。

执行结果比较长，然后我这里就不展示了

不同窗口的总结

image.png|697

优化建议以及原理

主要记忆流程：用户新问题 → 记忆组件提取历史对话 → 把“历史+新问题”拼起来 → 发给LLM → LLM生成回复 → 把“新问题+回复”存到记忆里 → 输出结果
主要的两个插件组合：

ChatMessageHistory：相当于“记忆笔记本”，负责具体的存和取操作。
RunnableWithMessageHistory：相当于“记忆调度员”，负责协调整个流程——在调用LLM前自动取历史，调用后自动存新对话。

3.2 外部行动层（Tool）：让AI能“动手”解决问题

Tool组件就是给AI装上“手和脚”，让它能通过调用外部工具，解决了它只能回答不能操作的问题。

Tool工具调用就是让AI学会“思考→行动→反馈”的循环，从简单的问答助手脱离成一个可以帮你干活的智能体：

思考：自己能不能回答，要不要调用一些工具
行动：生成工具调用指令
反馈：根据工具返回查询结果

主要是利用三个核心组件：

Tool（工具）：具体的“干活工具”，比如查询天气的工具、读文件的工具。每个工具都有明确的名称和描述（AI就是通过描述知道该用哪个工具的）；
Toolkit（工具包）：把相关的工具打包在一起，比如“文件操作工具包”里包含读文件、写文件、列目录三个工具；
Agent（智能体）：团队的“指挥官”，负责协调LLM和工具——判断要不要调用工具、用哪个工具、怎么处理工具返回的结果。

查天气

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os

# ======================
# 1. 环境
# ======================
load_dotenv()
API_KEY = os.getenv("API_KEY")
BASE_URL = os.getenv("BASE_URL")
llm = ChatOpenAI(
    api_key=API_KEY,
    base_url=BASE_URL,
    model="deepseek-chat",
    temperature=0.3,
)

# ======================
# 2. 工具
# ======================
@tool
def weather_query(city: str) -> str:
    """查询指定城市天气"""
    weather_data = {
        "北京": "北京今日天气：晴，-2~8℃",
        "上海": "上海今日天气：多云，5~12℃",
        "广州": "广州今日天气：小雨，18~25℃",
    }
    return weather_data.get(city, f"暂无 {city} 数据")

tools = [weather_query]

# ======================
# 3. 创建 Agent（开启 debug）
# ======================
agent = create_agent(
    model=llm,
    tools=tools,
    debug=True,  # 👈 打开过程打印
)

# ======================
# 4. 运行
# ======================
response = agent.invoke({
    "messages": [
        {"role": "user", "content": "北京今天的天气怎么样？"}
    ]
})

print("\n最终回答：")
print(response["messages"][-1].content)

这一份代码中采用了自定义工具（@tool装饰器）

输出：


最终回答：
北京今天晴，气温约 `-2~8℃`。出门注意早晚保暖。

@tool工具源码

(function)
def tool(
    *,
    description: str | None = None,          # 工具功能描述（Agent判断是否调用的核心依据）
    return_direct: bool = False,             # 是否直接返回工具结果（False：LLM整理后返回）
    args_schema: ArgsSchema | None = None,   # 参数校验模型（Pydantic BaseModel）
    infer_schema: bool = True,               # 是否自动从类型注解推导参数schema
    response_format: Literal['content', 'content_and_artifact'] = "content",  # 返回格式
    parse_docstring: bool = False,           # 是否解析docstring生成描述/参数信息
    error_on_invalid_docstring: bool = True  # docstring解析失败时是否抛异常
)

一些拓展工具

文件操作工具（FileManagementToolkit）

ReadFile：读取本地文件（txt、docx 等）
WriteFile：写入本地文件
ListDirectory：查看文件夹内容
DeleteFile（部分版本）：删除文件

章节总结

核心知识点

Memory组件：解决LLM“健忘”的问题，重点掌握三种常用组件（全量、窗口、摘要）的适用场景和用法；
Tool组件：给LLM装“手脚”，让它能调用外部工具，重点掌握内置工具的使用和自定义工具的规范；
组件组合：用LCEL实现模块流水线，重点掌握“记忆+工具+Agent”的组合思路，能构建带记忆、能行动的复杂应用。

后面的优化方向

记忆层：用 Redis/PostgreSQL 替代内存存储，实现记忆持久化；
工具层：增加权限控制、重试机制、异常捕获（如文件写入失败提示）；
性能层：缓存高频工具调用结果、限制记忆长度控制 Token 消耗；
体验层：优化工具调用话术，让回答更自然。

菜单

分享

lang架构的agent开发 3

3.1 Memory(状态管理层):让大模型具有记忆功能

全量记忆

窗口记忆

摘要记忆

不同窗口的总结

优化建议以及原理

3.2 外部行动层（Tool）：让AI能“动手”解决问题

查天气

@tool工具源码

一些拓展工具

章节总结

lang架构的agent开发 1

lang架构的agent开发 2

目录

jscpc2026

问答卡片

深度工作

lang架构的agent开发 3