LangChain Agent 入门与上下文压缩
LangChain Agent 入门与上下文压缩
AI Agent 工程实践教程 · 第 08 章
理解 Agent 推理循环、工具选择和长上下文压缩在问答系统中的作用。
第八天_LangChain Agent入门
第八天：LangChain Agent 入门
前面第七天，我们已经学习了 LangChain 的基础用法：
- 用
ChatOpenAI创建模型对象。 - 用
ChatPromptTemplate组织提示词。 - 用
StrOutputParser解析模型输出。 - 用 LCEL 把
prompt | model | parser串成一条链。 - 用
RunnableLambda、RunnableParallel把自己的函数接进链路。 - 用 LangChain Tool API 把业务能力声明成工具。
第七天最后，我们已经看到一个重要结论：
Tool 是可以被调用的能力。
Agent 是会根据问题选择工具的流程。
所以第八天不再重复普通 Chain 的写法。
这一节课要解决的是：
当 AI 可以使用多个工具时，能不能不让我们每次都手写“模型 -> 工具 -> 模型”的流程？
这就是 LangChain Agent 要解决的问题。
一句话：
Agent 帮我们自动跑工具调用流程，但工具本身和业务规则还是我们自己写。
1. 先复盘：工具调用到底做了什么
第五天和第七天都讲过工具调用。
这里再用一句话复盘：
工具调用不是模型真的执行 Python 函数。
模型只是返回：我想调用哪个工具，参数是什么。
比如学生问：
我明天有什么课？
模型可能返回：
[
{
"name": "get_student_day_schedule",
"args": {
"date": "2026-07-09"
},
}
]
这段结果的意思是：
模型认为应该调用 get_student_day_schedule 这个工具。
参数是 date=2026-07-09。
但是注意：
模型并没有真的执行这个函数。
真正执行函数的是我们的 Python 程序。
2. 不用 Agent 时，我们要自己写流程
第七天的 Tool API 示例，本质上是我们自己写工具调用流程。
大概流程是这样：
用户问题
↓
第一次调用模型
↓
模型返回 tool_calls
↓
程序读取 tool_calls
↓
程序执行对应工具
↓
程序把工具结果放回 messages
↓
第二次调用模型
↓
模型生成最终回答
如果写成简化代码，大概是这样：
from langchain_core.messages import HumanMessage, ToolMessage
messages = [
HumanMessage(content="我明天有什么课？明天是 2026-07-09。")
]
# 第一次调用模型：让模型判断要不要调用工具。
ai_message = model_with_tools.invoke(messages)
messages.append(ai_message)
for tool_call in ai_message.tool_calls:
tool_name = tool_call["name"]
tool_args = tool_call["args"]
tool_id = tool_call["id"]
selected_tool = tool_map[tool_name]
tool_result = selected_tool.invoke(tool_args)
messages.append(
ToolMessage(
content=str(tool_result),
tool_call_id=tool_id,
)
)
# 第二次调用模型：把工具结果交给模型，让模型组织最终回答。
final_message = model_with_tools.invoke(messages)
print(final_message.content)
这段代码能跑。
但是如果工具越来越多，问题就来了：
- 每个工具都要放到
tool_map。 - 每次都要判断有没有
tool_calls。 - 每次都要执行工具。
- 每次都要把工具结果包装成
ToolMessage。 - 有些问题可能需要多次工具调用。
- 后面想看中间过程、做日志、做异常处理，也会越来越复杂。
所以这里就出现了 Agent。
3. Agent 帮我们做什么
Agent 可以先理解成：
一个会帮我们自动跑工具调用流程的 AI 执行器。
不用 Agent：
我们自己写：
模型 -> 判断 tool_calls -> 执行工具 -> 回填工具结果 -> 再调模型
使用 Agent：
我们把模型、工具、系统提示词交给 Agent。
Agent 帮我们跑完整个流程。
也就是说，原来我们要手写的这段循环：
调用模型
↓
看模型有没有 tool_calls
↓
有 tool_calls 就执行工具
↓
把工具结果放回 messages
↓
再调用模型
↓
直到模型给出最终回答
现在可以交给 LangChain Agent。
但是要注意：
Agent 不是帮我们写业务代码。
它不会自动知道课程表在哪里。
它不会自动查数据库。
它不会自动完成权限校验。
这些还是我们自己写。
Agent 帮我们省掉的是：
工具调用流程代码。
不是：
业务逻辑代码。
4. 模型、工具、Agent 的分工
这三个概念一定要分清楚。
4.1 模型负责什么
模型负责判断和生成文本。
比如：
用户问：我明天有什么课？
模型判断：
这是课程安排问题，需要调用课程查询工具。
然后模型返回：
我要调用 get_student_day_schedule，参数是 date=2026-07-09。
模型本身不会真的查课表。
4.2 工具负责什么
工具负责执行一个具体能力。
比如：
@tool
def get_student_day_schedule(date: str) -> str:
"""查询当前登录学生在指定日期的课程安排，date 格式为 yyyy-MM-dd。"""
...
这个工具真正做的是：
根据日期查询课程安排。
真实项目里，它可能会调用 Java 内部接口，也可能会查询数据库。
4.3 Agent 负责什么
Agent 负责组织流程。
它会：
- 调用模型。
- 读取模型返回的
tool_calls。 - 执行对应工具。
- 把工具结果放回对话上下文。
- 再调用模型。
- 最后返回模型的最终回答。
所以最准确的理解是：
模型负责做决定。
工具负责办具体事情。
Agent 负责把整个流程跑起来。
5. 第一个 LangChain Agent
先看最小写法。
from langchain.agents import create_agent
agent = create_agent(
model=model,
tools=[get_student_day_schedule],
system_prompt="你是一个学生课程助手。回答要简洁、自然，适合学生理解。",
)
这段代码里先记住三个参数：
| 参数 | 含义 |
|---|---|
model |
用哪个大模型思考和生成回答 |
tools |
这个 Agent 可以使用哪些工具 |
system_prompt |
告诉 Agent 它的身份、任务和边界 |
创建好 Agent 以后，就可以调用：
result = agent.invoke({
"messages": [
{
"role": "user",
"content": "我明天有什么课？明天是 2026-07-09。",
}
]
})
返回结果里会有 messages。
最后一条通常就是最终回答：
final_message = result["messages"][-1]
print(final_message.content)
这里注意一点：
agent.invoke 不是只调用一次模型。
如果中间需要工具，它内部可能会调用多次模型。
这也是 Agent 和普通 model.invoke 的区别。
5.1 后面三个示例重点看什么
下面会用同一个 Agent 连续处理三个问题。
这三个问题不是随便选的。
它们分别对应 Agent 的三种常见处理方式：
| 示例 | 用户问题 | Agent 可能怎么处理 |
|---|---|---|
| 示例一 | 什么是大模型？ |
普通概念问题，直接回答，不调用工具 |
| 示例二 | 我明天有什么课？ |
课程安排问题，调用课程查询工具 |
| 示例三 | 我们课程里 RAG 怎么实现？ |
课程项目实现问题，调用知识库工具 |
学习时重点观察：
同一个 Agent，
面对不同问题，
会选择不同处理方式。
这就是 Agent 和普通固定 Chain 的区别。
5.2 示例一：普通问题不调用工具
先定义两个工具。
第一个工具：查课程安排。
from langchain_core.tools import tool
@tool
def get_student_day_schedule(date: str) -> str:
"""查询当前登录学生在指定日期的课程安排，date 格式为 yyyy-MM-dd。"""
if date == "2026-07-09":
return "2026-07-09 19:00-21:00 有《LangChain Agent 入门》课程。"
return f"{date} 暂未查询到课程安排。"
第二个工具：查询课程知识库。
@tool
def search_course_knowledge(question: str) -> str:
"""查询本课程项目的实现资料。只适合回答“我们课程里怎么实现”“项目代码怎么做”“课程资料里怎么讲”的问题。"""
if "RAG" in question or "rag" in question:
return (
"课程里的 RAG 实现流程是："
"先把课程资料切成 chunk，使用 embedding 模型转成向量，"
"再存入 Milvus。学生提问时，Python 服务会根据问题去 Milvus 检索相关 chunk，"
"然后把检索结果拼进 Prompt，最后交给大模型生成回答。"
)
return "知识库中暂时没有找到相关课程资料。"
然后创建 Agent：
from langchain.agents import create_agent
from test.langchain_demo.demo_model import build_demo_model
model = build_demo_model()
agent = create_agent(
model=model,
tools=[
get_student_day_schedule,
search_course_knowledge,
],
system_prompt=(
"你是一个学生课程助手。"
"如果学生只是问普通 AI 概念，比如“什么是大模型”“什么是 RAG”，直接回答，不要调用工具。"
"如果学生问课程安排，调用 get_student_day_schedule。"
"如果学生问本课程项目里的实现细节，比如“我们课程里 RAG 怎么实现”，调用 search_course_knowledge。"
"回答要简洁、自然，适合学生理解。"
),
)
现在问一个普通概念问题：
result = agent.invoke({
"messages": [
{
"role": "user",
"content": "什么是大模型？",
}
]
})
print(result["messages"][-1].content)
这个问题是普通概念解释。
Agent 可能不需要调用工具，直接让模型回答：
大模型可以先理解成一种参数规模很大、学过大量文本和知识的 AI 模型。它能根据输入内容生成回答、总结文本、写代码，也可以结合工具和知识库完成更复杂的任务。
这里可以这样理解：
Agent 不等于每次都调用工具。
Agent 会根据模型判断决定是否需要工具。
如果这里仍然问：
什么是 RAG？
也可能触发 search_course_knowledge。
因为工具说明里如果写了“适合回答 RAG 相关问题”，模型就可能认为这个问题应该查知识库。
所以实际使用时要把两个问题区分开：
什么是 RAG？
普通概念问题，直接回答。
我们课程里 RAG 怎么实现？
课程项目实现问题，调用知识库工具。
5.3 示例二：课程安排问题调用工具
再问一个业务问题：
result = agent.invoke({
"messages": [
{
"role": "user",
"content": "我明天有什么课？明天是 2026-07-09。",
}
]
})
print(result["messages"][-1].content)
这个问题不是普通知识解释。
它需要查课程安排。
Agent 内部大概会这样执行：
1. 调用模型，模型判断需要查课程。
2. 模型返回 tool_call：get_student_day_schedule(date="2026-07-09")。
3. Agent 执行 get_student_day_schedule。
4. 工具返回：2026-07-09 19:00-21:00 有《LangChain Agent 入门》课程。
5. Agent 把工具结果交回模型。
6. 模型组织最终回答。
最终回答可能是：
你明天晚上 19:00-21:00 有《LangChain Agent 入门》课程。
这里可以记住：
模型负责判断要查课。
工具负责返回课程数据。
Agent 负责把这个流程跑完。
5.4 示例三：课程实现问题调用知识库工具
再看第三个问题：
result = agent.invoke({
"messages": [
{
"role": "user",
"content": "我们课程里 RAG 怎么实现？",
}
]
})
print(result["messages"][-1].content)
这个问题和课程项目实现有关。
它不是问通用概念，而是在问：
我们这个课程项目里具体怎么做 RAG？
所以 Agent 可以调用知识库查询工具：
search_course_knowledge(question="我们课程里 RAG 怎么实现？")
工具返回课程资料以后，模型再组织回答：
我们课程里的 RAG 大致是这样实现的：先把课程资料切成 chunk，然后用 embedding 模型转成向量，存入 Milvus。学生提问时，Python 服务根据问题去 Milvus 检索相关资料，再把检索结果拼进 Prompt，最后交给大模型生成回答。
这个例子能体现 Agent 的价值：
同样都提到 RAG：
问“什么是 RAG？”是普通概念解释，可以直接回答。
问“我们课程里 RAG 怎么实现？”是在问课程项目实现，更适合查课程知识库。
5.5 完整 Demo
第八天的 Agent 示例已经单独放在项目里的 test/agent_demo 包下面。
建议按这个顺序看：
01_create_agent_demo.py 创建 Agent，并查看它有哪些工具
02_direct_answer_demo.py 普通概念问题：直接回答
03_schedule_tool_demo.py 课程安排问题：调用课程查询工具
04_knowledge_tool_demo.py 课程实现问题：调用知识库工具
05_compare_three_questions_demo.py 对比三类问题的最终回答
06_messages_process_demo.py 打印 Agent 中间消息，看清工具调用过程
07_multi_step_tools_demo.py 多步工具调用：先查课程，再查课程详情
08_short_term_memory_demo.py 短期记忆：同一个 thread_id 会保留对话历史
完整的三类问题对比代码在：
test/agent_demo/05_compare_three_questions_demo.py
from test.agent_demo.agent_factory import build_course_agent
from test.agent_demo.message_printer import print_final_answer
def main():
agent = build_course_agent()
questions = [
"什么是大模型？",
"我明天有什么课？明天是 2026-07-09。",
"我们课程里 RAG 怎么实现？",
]
for question in questions:
print("=" * 60)
print("学生问题：", question)
result = agent.invoke({
"messages": [
{
"role": "user",
"content": question,
}
]
})
print("最终回答：")
print_final_answer(result)
if __name__ == "__main__":
main()
运行：
uv run python test/agent_demo/05_compare_three_questions_demo.py
5.6 多步工具调用
Agent 不一定只调用一次工具。
有些问题需要先调用一个工具，拿到结果以后，再决定要不要调用第二个工具。
比如这个问题：
我明天这节课主要讲什么？明天是 2026-07-09。
Agent 可以这样处理：
第 1 步：调用 get_student_day_schedule，先查明天是哪节课。
第 2 步：工具返回《LangChain Agent 入门》。
第 3 步：模型继续判断：现在知道课程名称了，还需要查课程详情。
第 4 步：调用 get_lesson_detail，查询这节课主要讲什么。
第 5 步：模型根据两个工具结果生成最终回答。
这就是：
模型 -> 工具 -> 模型 -> 工具 -> 模型
对应示例在：
test/agent_demo/07_multi_step_tools_demo.py
运行：
uv run python test/agent_demo/07_multi_step_tools_demo.py
这个例子能看出：
Agent 每次拿到工具结果后，还会把结果交回模型。
模型可以继续判断下一步是回答，还是继续调用其他工具。
6. 怎么看 Agent 有没有调用工具
直接用 invoke 时，我们通常只看最终结果。
如果想看清楚 Agent 中间做了什么，可以把返回的 messages 打印出来。
先看一段更容易理解的代码：
result = agent.invoke({
"messages": [
{
"role": "user",
"content": "我明天有什么课？明天是 2026-07-09。",
}
]
})
for message in result["messages"]:
print("=" * 40)
print("消息角色：", message.type)
if message.content:
print("消息内容：")
print(message.content)
if getattr(message, "tool_calls", None):
print("工具调用：")
print(message.tool_calls)
这段代码没有引入新的 Agent 能力。
它只是把 Agent 执行后的所有消息打印出来。
如果 Agent 调用了工具，messages 里大概会出现这样的顺序：
第 1 条：human
用户问题：我明天有什么课？
第 2 条：ai
模型没有直接回答，而是生成 tool_calls。
意思是：我想调用课程查询工具。
第 3 条：tool
工具执行后的结果。
比如：2026-07-09 19:00-21:00 有《LangChain Agent 入门》课程。
第 4 条：ai
模型根据工具结果，生成最终回答。
所以判断有没有调用工具，可以重点看：
中间有没有带 tool_calls 的 ai 消息。
中间有没有 tool 消息。
如果想打印得更清楚，可以写成这样：
for message in result["messages"]:
print("=" * 40)
message_type = message.type
if message_type == "human":
print("用户问题：")
print(message.content)
elif message_type == "ai":
if getattr(message, "tool_calls", None):
print("模型决定调用工具：")
for tool_call in message.tool_calls:
print("工具名：", tool_call["name"])
print("参数：", tool_call["args"])
else:
print("模型回答：")
print(message.content)
elif message_type == "tool":
print("工具返回结果：")
print(message.content)
如果 Agent 决定调用课程查询工具，通常可以看到类似输出：
用户问题：
我明天有什么课？明天是 2026-07-09。
模型决定调用工具：
工具名： get_student_day_schedule
参数： {'date': '2026-07-09'}
工具返回结果：
2026-07-09 19:00-21:00 有《LangChain Agent 入门》课程。
模型回答：
你明天晚上 19:00-21:00 有《LangChain Agent 入门》课程。
也可以使用 stream 一边运行一边看中间过程。
但刚开始学习 Agent 时，先用 invoke 跑完，再打印 messages 会更容易理解。
for chunk in agent.stream(
{
"messages": [
{
"role": "user",
"content": "我明天有什么课？明天是 2026-07-09。",
}
]
},
stream_mode="values",
):
latest_message = chunk["messages"][-1]
print(type(latest_message).__name__, latest_message)
这一步的重点不是记住打印代码。
它能帮助我们看清楚：
Agent 不是神秘地直接给出答案。
它中间真的经历了工具选择和工具执行。
7. Agent 的短期记忆：checkpointer
在继续看 Agent 内部流程之前，还可以补一个很常见的问题：
Agent 能不能记住前面聊过什么？
可以。
LangChain Agent 可以通过 checkpointer 保存短期记忆。
先看一个普通对话问题。
用户先说：
我叫小明。
然后用户再问：
我叫什么？
如果我们每次调用 Agent 时，都只传当前这一句话：
result = agent.invoke({
"messages": [
{
"role": "user",
"content": "我叫什么？",
}
]
})
模型其实不知道“小明”这个信息。
因为这一次请求里只有：
我叫什么？
前面那句“我叫小明”没有传进去。
所以，如果想让 Agent 记住同一段对话里的上下文，就需要让它保存历史消息。
这就是 checkpointer 的作用。
可以先把它理解成：
checkpointer 用来保存同一个对话里的历史消息。
7.1 为什么这里会出现 LangGraph
代码里会看到这个导入：
from langgraph.checkpoint.memory import InMemorySaver
注意：
InMemorySaver 确实来自 LangGraph。
这是正常的。
因为新版 LangChain 的 Agent 底层已经使用了 LangGraph 的能力。
这里可以先这样理解：
我们创建 Agent 用的是 LangChain。
Agent 的短期记忆保存器来自 LangGraph。
这一节不用展开讲 LangGraph 的节点、边、状态图。
现在只需要知道：
InMemorySaver 可以把同一个 thread_id 里的消息临时保存起来。
7.2 给 Agent 加上短期记忆
先创建一个保存器：
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
然后创建 Agent 时传进去：
from langchain.agents import create_agent
agent = create_agent(
model=model,
tools=[],
system_prompt="你是一个学生课程助手。",
checkpointer=checkpointer,
)
这样 Agent 就有了保存短期对话历史的能力。
但是还差一个关键点：
我们要告诉 Agent，当前是哪一段对话。
这要靠 thread_id。
config = {
"configurable": {
"thread_id": "student-1001"
}
}
可以把 thread_id 理解成一次对话的编号。
同一个 thread_id，表示还在同一段对话里。
不同的 thread_id，表示是不同的对话。
7.3 同一个 thread_id 会记住上下文
第一次调用：
agent.invoke(
{
"messages": [
{
"role": "user",
"content": "我叫小明。",
}
]
},
config=config,
)
第二次调用：
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "我叫什么？",
}
]
},
config=config,
)
print(result["messages"][-1].content)
因为两次调用使用的是同一个 thread_id：
student-1001
所以第二次问“我叫什么”时，Agent 可以看到前面保存过的消息。
它就有机会回答：
你叫小明。
7.4 换一个 thread_id 就不是同一段对话
如果换成另一个 thread_id：
other_config = {
"configurable": {
"thread_id": "student-2002"
}
}
然后再问：
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "我叫什么？",
}
]
},
config=other_config,
)
print(result["messages"][-1].content)
这个时候，Agent 不应该知道“小明”。
因为 student-2002 是另一段对话。
这就像两个不同的聊天窗口：
student-1001 这段对话里，用户说过“我叫小明”。
student-2002 这段对话里，用户没有说过这句话。
7.5 checkpointer 记住的是什么
checkpointer 主要保存的是 Agent 的运行状态。
刚开始可以重点理解为：
它保存了这段对话里的 messages。
也就是类似这样的消息历史：
human: 我叫小明。
ai: 好的，我记住了。
human: 我叫什么？
ai: 你叫小明。
如果这段对话中间发生过工具调用，消息里也可能包含：
ai: 准备调用工具
tool: 工具返回结果
ai: 根据工具结果回答
所以 checkpointer 不是让模型真的“脑子里永久记住了用户”。
它只是帮 Agent 把同一段对话的历史保存下来。
下一次同一个 thread_id 再来时，Agent 可以继续接着这段历史往下聊。
7.6 InMemorySaver 底层怎么保存
InMemorySaver 可以先理解成 Agent 的临时存档本。
每次 Agent 运行时，都会产生一份当前状态。
这份状态叫 checkpoint。
checkpoint 可以理解成一次存档。
比如第一次对话：
用户：我叫小明。
Agent 运行结束后，会保存一份状态：
checkpoint_1
里面记录：这段对话里，用户说过“我叫小明”。
第二次同一个 thread_id 再问：
用户：我叫什么？
Agent 会先根据 thread_id 找到上一次保存的状态。
然后把新的问题接到历史后面，再交给模型回答。
所以它才能回答：
你叫小明。
InMemorySaver 底层不是数据库，也不是文件。
它是保存在 Python 程序内存里的。
可以简单理解成：
{
"student-1001": {
"checkpoint_1": "第一次对话后的状态",
"checkpoint_2": "第二次对话后的状态",
},
"student-2002": {
"checkpoint_1": "另一段对话的状态",
}
}
所以它有三个特点：
同一个 thread_id，可以找到前面的状态。
不同的 thread_id，互相看不到对方的状态。
程序停止后，内存释放，状态也就没了。
一句话总结：
InMemorySaver 就是用内存字典保存 Agent 的临时运行状态。
thread_id 用来区分是哪一段对话。
checkpoint 是每次运行后保存下来的状态快照。
7.7 InMemorySaver 保存在哪里
InMemorySaver 里的 memory 表示内存。
所以它的意思是：
把对话历史先保存在当前 Python 程序的内存里。
内存可以先理解成：
程序运行时临时使用的一块空间。
所以 InMemorySaver 很适合刚开始学习。
因为它不用连接数据库，也不用额外配置。
但是要注意：
它保存的数据不是永久的。
程序一停，内存里的数据就没了。
比如：
重启 Python 程序
重启服务器
服务崩溃后重新启动
这些情况都会导致 InMemorySaver 里的数据丢失。
如果是真正上线的系统，要长期保存用户对话历史，通常会把数据保存到数据库、Redis 或其他存储里。
7.8 历史消息会不会无限变长
如果一直使用同一个 thread_id 聊下去，历史消息确实会越来越多。
比如：
第 1 轮：我叫小明。
第 2 轮：我明天有什么课？
第 3 轮：刚才我叫什么？
第 4 轮：刚才那节课讲什么？
...
这些消息都会被保存到这个 thread_id 对应的历史里。
但是这不等于模型每次都能无限读完所有历史。
因为模型有一个限制：
上下文长度
上下文长度可以先理解成：
模型一次最多能看多少内容。
如果历史消息太多，可能会出现这些问题：
请求内容太长，直接报错。
模型只能看到一部分历史。
回答速度变慢。
调用成本变高。
所以不要把 checkpointer 理解成：
一直聊，Agent 会自动把所有历史整理好，并且永远记住。
更准确的理解是：
checkpointer 负责保存历史。
历史太长以后，怎么裁剪、怎么总结、哪些内容要留下来，需要我们自己设计。
比如真正的系统里，可能会这样处理：
只保留最近几轮对话。
把前面的对话总结成一小段摘要。
把重要信息单独保存到数据库。
所以短期记忆不是无限记忆。
它只是让 Agent 能在同一段对话里，接着前面的内容继续回答。
7.9 短期记忆和长期记忆不是一回事
这里讲的是短期记忆。
它解决的是：
同一段对话里，Agent 能不能接着前面的话继续聊。
比如：
用户：我叫小明。
用户：我明天有什么课？
用户：那我叫什么？
只要是同一个 thread_id，Agent 就可以依赖前面的消息继续回答。
长期记忆是另一件事。
长期记忆更像：
系统长期记住某个学生的偏好、学习情况、历史行为。
比如：
这个学生 Python 基础比较弱。
这个学生更喜欢看示例代码。
这个学生上次问过 RAG 的 Milvus 实现。
这类内容一般不只是保存一段聊天消息，而是要设计专门的存储、更新和读取规则。
所以第八天先掌握：
checkpointer = 短期记忆
thread_id = 区分是哪一段对话
InMemorySaver = 把数据临时保存在内存里的保存器
历史消息太长时，会受到模型上下文长度限制
对应的示例代码在：
test/agent_demo/08_short_term_memory_demo.py
运行方式：
uv run python test/agent_demo/08_short_term_memory_demo.py
长期记忆后面再结合 LangGraph 或项目里的数据库设计单独讲。
8. Agent 内部可以怎么理解
虽然我们使用 Agent 时只写：
result = agent.invoke({
"messages": [
{
"role": "user",
"content": "我明天有什么课？",
}
]
})
但它内部可以粗略理解成这样的循环：
messages = [
{
"role": "user",
"content": "我明天有什么课？",
}
]
while True:
ai_message = model.invoke(messages)
if not ai_message.tool_calls:
print(ai_message.content)
break
messages.append(ai_message)
for tool_call in ai_message.tool_calls:
tool_name = tool_call["name"]
tool_args = tool_call["args"]
tool_result = run_tool(tool_name, tool_args)
messages.append({
"role": "tool",
"tool_call_id": tool_call["id"],
"content": tool_result,
})
这段伪代码就是 Agent 的核心思想：
反复调用模型。
如果模型要工具，就执行工具。
如果模型不要工具，就输出最终回答。
所以 Agent 不是一个新模型。
它是一个流程。
9. Agent 和 Chain 的区别
第七天讲的 Chain 更像固定流水线。
比如：
chain = prompt | model | parser
它的流程是固定的：
Prompt -> Model -> Parser
无论用户问什么，它都按这个顺序执行。
Agent 不一样。
Agent 的流程会根据模型返回结果变化：
用户问题
↓
模型判断
↓
不需要工具 -> 直接回答
↓
需要工具 -> 调用工具 -> 再交给模型 -> 最终回答
所以可以这样对比：
| 对比项 | Chain | Agent |
|---|---|---|
| 流程 | 固定 | 会根据问题变化 |
| 是否使用工具 | 通常由程序写死 | 模型判断要不要用 |
| 适合场景 | 固定问答、固定 RAG 流程 | 多工具、多步骤、不确定是否需要工具 |
| 控制难度 | 更好控制 | 更灵活，但要注意边界 |
不要觉得 Agent 一定比 Chain 高级。
真实项目里要看场景：
如果流程非常固定，用 Chain 就够了。
如果工具很多，而且要让模型自己判断下一步，可以考虑 Agent。
10. Agent 和我们项目里的 ToolChatAgent
我们项目里之前已经有手写的工具调用流程。
比如：
ChatService 判断是否需要工具
↓
ToolChatAgent 负责工具调用流程
↓
ChatToolRegistry 根据工具名执行具体工具
↓
工具结果回到模型生成最终回答
这和 LangChain Agent 的思路很像。
区别是：
手写 ToolChatAgent：
流程由我们自己控制，代码更贴近项目。
LangChain Agent：
工具调用循环由 LangChain 帮我们封装，写起来更简洁。
所以不是说用了 LangChain Agent，就必须把项目里所有代码都换掉。
更稳的学习方式是：
先用这个小例子理解 Agent 原理。
再判断项目里哪些地方适合引入 LangChain Agent。
11. 工具设计仍然要注意安全
Agent 能不能用工具，不是它自己决定的。
是我们在代码里给它哪些工具，它才有哪些工具。
例如：
agent = create_agent(
model=model,
tools=[
get_student_day_schedule,
search_course_knowledge,
],
system_prompt="你是一个学生课程助手。",
)
这表示：
这个 Agent 只能使用这两个工具。
它不能凭空调用一个删除订单工具。
除非你把删除订单工具也放进 tools。
所以安全边界的第一条是：
只给 Agent 开放必要工具。
第二条：
不要把敏感身份参数交给模型决定。
比如查课程工具，不建议这样设计：
@tool
def get_student_day_schedule(student_id: int, date: str) -> str:
"""查询某个学生某一天的课程安排。"""
...
因为用户可能说：
帮我查 studentId=10086 明天的课。
模型可能就把 10086 当成参数传进去。
更好的设计是：
@tool
def get_student_day_schedule(date: str) -> str:
"""查询当前登录学生在指定日期的课程安排，date 格式为 yyyy-MM-dd。"""
...
当前学生是谁，应该来自 Java 鉴权后的上下文。
不是来自模型。
第三条：
高风险操作不要直接交给 Agent 自动执行。
例如：
- 删除数据。
- 发起退款。
- 修改支付状态。
- 群发通知。
- 修改学生重要资料。
这类操作即使以后要接入 Agent，也应该增加人工确认、权限校验和审计日志。
12. 什么时候适合用 Agent
适合 Agent 的场景：
- 工具不止一个。
- 用户问题类型不固定。
- 需要模型判断要不要调用工具。
- 工具结果回来后，还要让模型组织成自然语言。
- 有些问题可以直接回答，有些问题必须查业务数据。
例如学生端 AI 问答：
什么是 RAG？
我明天有什么课？
我们课程里 RAG 怎么实现？
我这个订单支付了吗？
我有哪些待完成作业？
这些问题背后可能需要不同处理方式。
Agent 就比较适合。
不一定适合 Agent 的场景：
- 流程非常固定。
- 不需要模型判断分支。
- 对结果确定性要求非常高。
- 每一步都必须严格按业务状态机执行。
比如支付回调、退款状态流转、订单超时处理，这类核心业务流程不应该交给 Agent 随意决定。
13. 今天要带走什么
第八天最重要的不是记住 API。
而是理解 Agent 到底帮我们做了什么。
可以记住这几句话：
1. Tool 是一个可以被 AI 使用的函数。
2. 模型不会真的执行工具，它只会返回 tool_calls。
3. 不用 Agent 时，我们要自己写“模型 -> 工具 -> 模型”的流程。
4. LangChain Agent 可以帮我们自动跑这套工具调用流程。
5. Agent 省掉的是流程代码，不是业务代码。
6. checkpointer 可以让 Agent 记住同一个 thread_id 里的短期对话历史。
7. 工具本身、权限校验、安全边界，还是程序员自己负责。
最后用一句话总结：
Agent 不是让 AI 变成万能系统，而是让 AI 在我们提供的工具范围内，更自动地完成一次问答流程。
第八天_问答上下文压缩设计
第八天：问答上下文压缩设计
前面讲 checkpointer 时，我们知道了一个问题：
如果一直把历史对话都塞给模型，消息会越来越长。
但是模型一次能看的内容是有限的。
这个限制叫：
上下文长度
所以在真实项目里，不能每次都把一个用户的所有历史问答都传给大模型。
更常见的做法是：
数据库保存完整历史。
调用模型时，只传“摘要 + 最近未压缩的对话 + 当前问题”。
这就是问答上下文压缩。
1. 为什么要做上下文压缩
假设一个学生一直和 AI 聊天：
第 1 轮：什么是 RAG？
第 2 轮：embedding 是什么？
第 3 轮：Milvus 是什么？
第 4 轮：LangChain Agent 是什么？
...
第 100 轮：我们项目里这个怎么实现？
如果每次都把前 100 轮对话全部发给大模型，会有几个问题：
1. 请求内容越来越长。
2. 调用成本越来越高。
3. 响应速度越来越慢。
4. 超过模型上下文长度后，可能直接报错。
5. 历史太多时，模型也更容易抓不住重点。
所以我们要把比较早的对话压缩成一段摘要。
比如原始历史是：
用户：什么是 RAG？
AI：RAG 是检索增强生成。
用户：我们项目里 RAG 怎么做？
AI：先切分资料，再做 embedding，存入 Milvus，提问时检索相关 chunk。
用户：Milvus 是干什么的？
AI：Milvus 是向量数据库，用来做相似度检索。
压缩后可以变成：
学生正在学习 RAG。已经了解：RAG 是检索增强生成；项目里会把资料切分成 chunk，
用 embedding 转成向量并存入 Milvus；提问时从 Milvus 检索相关内容再交给大模型回答。
后面再问问题时，就不用把这几轮原始对话都传给模型了。
只传这段摘要就可以保留主要背景。
2. 核心思路
项目里已经会保存每次问答记录。
所以我们不需要把 LangChain 的 InMemorySaver 直接接到项目里。
项目可以自己做上下文管理：
完整问答历史：保存到数据库。
压缩后的背景：保存到会话摘要字段。
每次问答上下文：摘要 + 未压缩消息 + 当前问题。
整体流程可以先看这张图：
flowchart LR
Q["用户提问"] --> SaveUser["保存用户消息"]
SaveUser --> Context["读取上下文"]
Context --> Prompt["组装 Prompt"]
Prompt --> Model["大模型回答"]
Model --> SaveAi["保存 AI 回答"]
SaveAi --> Check{"达到压缩阈值？"}
Check -- "否" --> Done["本次结束"]
Check -- "是" --> Async["提交异步压缩任务"]
Async --> Done
subgraph ContextData["上下文来源"]
Summary["会话摘要 summary"]
Recent["未压缩消息"]
end
Summary -.-> Context
Recent -.-> Context
subgraph CompressJob["异步压缩任务"]
Lock["获取会话压缩锁"]
Pick["选择旧消息"]
Merge["旧摘要 + 旧消息"]
NewSummary["生成新摘要"]
Mark["标记消息已压缩"]
end
Async -.-> Lock
Lock --> Pick
Pick --> Merge
Merge --> NewSummary
NewSummary --> Mark
classDef main fill:#E8F3FF,stroke:#5B8DEF,color:#1F2D3D,stroke-width:1px;
classDef data fill:#F7F0FF,stroke:#A66CFF,color:#2F2345,stroke-width:1px;
classDef job fill:#EAF8EF,stroke:#4CAF73,color:#1F3D2A,stroke-width:1px;
classDef decision fill:#FFF4D6,stroke:#E6A23C,color:#3D2F14,stroke-width:1px;
class Q,SaveUser,Context,Prompt,Model,SaveAi,Done main;
class Summary,Recent data;
class Pick,Merge,NewSummary,Mark,Async job;
class Check decision;
这张图里要注意：
回答用户问题是主流程。
压缩历史消息是异步流程。
也就是说，压缩不应该阻塞用户当前这次问答。
3. 数据库可以怎么设计
这里不一定要求表字段完全这样写。
重点是理解需要保存哪些信息。
3.1 会话表
比如有一张会话表：
ai_chat_session
可以保存：
id 会话 ID
student_id 学生 ID
summary 当前会话摘要
last_compressed_message_id 最后压缩到哪一条消息
summary_token_count 摘要的大概 token 数
updated_at 更新时间
其中最重要的是：
summary
last_compressed_message_id
summary 保存压缩后的背景。
last_compressed_message_id 表示：
这条消息以及这条消息之前的内容，已经被压缩进 summary 里了。
3.2 消息表
比如有一张消息表：
ai_chat_message
可以保存：
id 消息 ID
session_id 会话 ID
role user / assistant
content 消息内容
token_count 这条消息的大概 token 数
compressed 是否已经压缩
created_at 创建时间
compressed 可以先用简单的布尔值：
0 = 未压缩
1 = 已压缩
也可以不用单独查 compressed 字段，而是根据：
message.id <= last_compressed_message_id
来判断这条消息是否已经压缩过。
4. 压缩阈值放到 dataConfig
这些数字不要写死在代码里。
因为不同模型、不同课程、不同费用预算下，合适的阈值可能不一样。
可以放到项目的动态配置 dataConfig 里。
例如：
ai.chat.context.compress.enabled=true
ai.chat.context.compress.messageThreshold=20
ai.chat.context.compress.keepRecentMessageCount=8
ai.chat.context.compress.maxCompressMessageCount=12
ai.chat.context.compress.tokenThreshold=6000
含义可以这样理解：
enabled
是否开启上下文压缩。
messageThreshold
未压缩消息超过多少条后，触发压缩。
keepRecentMessageCount
最近多少条消息保留原文，不压缩。
maxCompressMessageCount
一次最多压缩多少条旧消息。
tokenThreshold
未压缩消息 token 数超过多少后，也可以触发压缩。
比如配置是：
messageThreshold=20
keepRecentMessageCount=8
maxCompressMessageCount=12
意思是：
未压缩消息超过 20 条时，可以触发压缩。
压缩时最多压缩前 12 条。
最近 8 条继续保留原文。
为什么要保留最近几条原文？
因为最近的对话通常细节最多。
摘要适合保存长期背景。
最近原文适合保留具体上下文。
5. 每次问答时怎么查询上下文
每次用户提问时，不要查所有历史消息。
应该查：
1. 当前会话的 summary
2. 当前会话里未压缩的消息
3. 当前用户的新问题
最终给大模型的上下文可以理解成这样：
flowchart TD
DB["数据库完整历史"] --> Old["已压缩旧消息"]
DB --> Recent["未压缩最近消息"]
Old --> Summary["会话摘要 summary"]
subgraph PromptBox["本次发送给模型的 Prompt"]
System["系统提示词"]
SummaryIn["历史摘要"]
RecentIn["最近原文"]
Question["当前问题"]
end
Summary --> SummaryIn
Recent --> RecentIn
User["用户本次提问"] --> Question
PromptBox --> Model["大模型"]
classDef source fill:#F5F7FA,stroke:#9AA4B2,color:#1F2D3D,stroke-width:1px;
classDef summary fill:#F7F0FF,stroke:#A66CFF,color:#2F2345,stroke-width:1px;
classDef prompt fill:#E8F3FF,stroke:#5B8DEF,color:#1F2D3D,stroke-width:1px;
classDef model fill:#EAF8EF,stroke:#4CAF73,color:#1F3D2A,stroke-width:1px;
class DB,Old,Recent,User source;
class Summary,SummaryIn summary;
class System,RecentIn,Question prompt;
class Model model;
这里的重点是：
已压缩旧消息不再逐条放进 Prompt。
它们已经变成 summary。
未压缩的最近消息继续保留原文。
可以理解成：
select summary
from ai_chat_session
where id = ?
然后查未压缩消息：
select *
from ai_chat_message
where session_id = ?
and id > last_compressed_message_id
order by id asc
如果使用 compressed 字段，也可以这样理解：
select *
from ai_chat_message
where session_id = ?
and compressed = 0
order by id asc
最后组装给大模型的内容大概是：
系统提示词
【历史摘要】
{summary}
【最近对话】
{未压缩的历史消息}
【当前问题】
{用户这次的问题}
这样模型既能看到长期背景，也能看到最近细节。
6. 已经有摘要后，新的对话怎么压缩
这是上下文压缩里最重要的一点。
如果已经有摘要了，后面又产生了新的对话，需要再次压缩。
这个时候不要把旧历史全部查出来重新总结。
也不要只总结本次新增对话。
应该做：
旧摘要 + 本次要压缩的新消息 -> 新摘要
这叫增量摘要。
可以先看这张图：
flowchart LR
Empty["空摘要"] --> Merge1["合并第 1 批旧消息"]
Msg1["第 1 批旧消息"] --> Merge1
Merge1 --> S1["summary_v1"]
S1 --> Merge2["合并第 2 批旧消息"]
Msg2["第 2 批旧消息"] --> Merge2
Merge2 --> S2["summary_v2"]
S2 --> Merge3["合并第 3 批旧消息"]
Msg3["第 3 批旧消息"] --> Merge3
Merge3 --> S3["summary_v3"]
classDef summary fill:#F7F0FF,stroke:#A66CFF,color:#2F2345,stroke-width:1px;
classDef message fill:#FFF4D6,stroke:#E6A23C,color:#3D2F14,stroke-width:1px;
classDef merge fill:#E8F3FF,stroke:#5B8DEF,color:#1F2D3D,stroke-width:1px;
class Empty,S1,S2,S3 summary;
class Msg1,Msg2,Msg3 message;
class Merge1,Merge2,Merge3 merge;
每次压缩都不是重新处理全部历史。
而是：
上一版摘要 + 新的一批旧消息 -> 下一版摘要
比如第一次压缩：
空摘要 + 第 1 批旧消息 -> summary_v1
第二次压缩：
summary_v1 + 第 2 批旧消息 -> summary_v2
第三次压缩：
summary_v2 + 第 3 批旧消息 -> summary_v3
也就是说，摘要会不断更新。
它不是只生成一次。
6.1 一个例子
已有摘要：
学生叫小明，正在学习 RAG。已经了解 embedding 和 Milvus 的基本作用。
本次新增对话：
用户：LangChain Agent 是什么？
AI：Agent 可以根据问题决定是否调用工具。
用户：checkpointer 是什么？
AI：checkpointer 用来保存同一个 thread_id 的短期对话状态。
新的摘要应该变成：
学生叫小明，正在学习 RAG 和 LangChain Agent。
已经了解 embedding、Milvus、Agent 工具调用流程，
以及 checkpointer 可以保存同一个 thread_id 的短期对话状态。
这样旧摘要里的重要内容不会丢。
新对话里的重要内容也会被合并进去。
7. 摘要 Prompt 可以怎么写
压缩摘要时，最好使用一个专门的 Prompt。
不要只写：
帮我总结一下。
这样太模糊。
可以写得明确一点：
你负责维护一段学生 AI 问答的对话摘要。
下面有两部分内容：
【已有摘要】
{old_summary}
【本次新增对话】
{messages_to_compress}
请把它们合并成一个新的摘要。
要求：
1. 保留用户明确说过的个人信息、学习目标、学习偏好。
2. 保留已经确认过的重要技术结论。
3. 保留用户当前正在关注的问题。
4. 如果新增对话修正了已有摘要，以新增对话为准。
5. 删除寒暄、重复解释和无关内容。
6. 不要编造没有出现过的信息。
7. 用简洁中文输出。
这里要注意一句：
不是重新总结本次新增对话。
而是把已有摘要和新增对话合并成新的摘要。
8. Java 异步压缩流程
压缩不建议阻塞用户当前问答。
也就是说，用户问问题时，主流程应该尽快返回回答。
压缩可以放到异步任务里做。
主流程：
1. 用户提问。
2. 保存用户消息。
3. 查询 summary + 未压缩消息。
4. 调用大模型生成回答。
5. 保存 AI 回答。
6. 判断是否达到压缩阈值。
7. 如果达到阈值，提交异步压缩任务。
8. 当前问答正常返回给用户。
异步任务：
1. 读取 dataConfig 压缩配置。
2. 获取当前会话的压缩锁。
3. 如果没有拿到锁，说明已经有压缩任务在跑，直接结束。
4. 查询当前会话 summary。
5. 查询未压缩消息。
6. 保留最近 keepRecentMessageCount 条原文。
7. 从更早的消息里选出本次要压缩的消息。
8. 用 old_summary + messages_to_compress 调用大模型生成 new_summary。
9. 更新会话 summary。
10. 更新 last_compressed_message_id。
11. 将本次压缩的消息标记为 compressed=1。
12. 释放压缩锁。
9. 异步压缩为什么要加锁
异步压缩一定要考虑并发问题。
因为用户可能连续快速提问。
比如：
第 21 条消息保存后，触发一次压缩任务。
第 22 条消息保存后，又触发一次压缩任务。
如果两个压缩任务同时处理同一个会话，可能会出现问题：
重复压缩同一批消息。
后完成的任务覆盖先完成的摘要。
last_compressed_message_id 更新错。
一部分消息被错误标记为已压缩。
所以压缩应该加一把会话级别的锁。
可以理解成：
同一个 session_id，同一时间只允许一个压缩任务执行。
不同会话之间不影响。
比如：
session_1001 可以压缩。
session_1002 也可以同时压缩。
但是 session_1001 不能同时跑两个压缩任务。
9.1 锁可以怎么做
如果项目是单机部署，可以先用数据库字段做一个简单锁。
比如会话表里加：
compressing
含义是：
0 = 当前没有压缩任务
1 = 当前正在压缩
抢锁时不要先查再改。
要用一条带条件的更新语句。
可以理解成：
update ai_chat_session
set compressing = 1
where id = ?
and compressing = 0
如果这条 SQL 影响行数是 1：
说明抢锁成功，可以开始压缩。
如果影响行数是 0：
说明已经有其他压缩任务在跑，这次任务直接结束。
压缩完成后，再释放锁：
update ai_chat_session
set compressing = 0
where id = ?
如果压缩失败，也要释放锁。
所以代码里要注意：
try {
执行压缩
} finally {
释放锁
}
如果项目后面是多实例部署，也可以用 Redis 分布式锁。
但刚开始实现时，用数据库字段做会话级锁就够理解这个思路了。
10. 需不需要 COMPRESSING 状态
可以先不需要。
简单版本可以只用：
compressed = 0
compressed = 1
也就是：
未压缩
已压缩
并发问题交给会话级压缩锁处理。
也就是：
消息只关心有没有被压缩。
会话负责控制当前有没有压缩任务正在执行。
这样消息状态更简单，学生也更容易理解。
11. 压缩时要注意什么
11.1 不要删除原始消息
压缩不是删除。
数据库里仍然保留完整原始消息。
只是后续调用模型时，不再把已经压缩过的消息逐条查出来。
这样以后如果要做审计、问题排查、聊天记录展示，原始数据还在。
11.2 不要每次都重新总结全部历史
如果每次压缩都查全部历史重新总结，会有几个问题：
成本高。
速度慢。
摘要容易漂移。
实现也更重。
所以更推荐：
旧摘要 + 本次新增要压缩的消息 -> 新摘要
11.3 最近几轮尽量保留原文
摘要适合保存长期背景。
但最近几轮对话通常有很多细节。
比如用户刚刚说：
我说的不是 LangChain memory，我说的是我们项目自己的问答记录。
这种细节如果只靠摘要，可能会丢失语气和上下文。
所以最近几轮最好继续保留原文。
11.4 摘要不能编造
压缩摘要也是大模型生成的。
大模型可能会把没有出现过的信息写进去。
所以摘要 Prompt 里一定要强调：
不要编造没有出现过的信息。
不确定的内容不要写成确定结论。
12. 最终效果
做完上下文压缩以后，每次问答给模型的内容会从：
全部历史消息 + 当前问题
变成：
历史摘要 + 最近未压缩消息 + 当前问题
这样可以：
减少上下文长度。
降低调用成本。
提升响应速度。
保留长期背景。
保留最近细节。
避免超过模型上下文限制。
一句话总结：
问答上下文压缩不是把历史删掉，而是把旧历史变成摘要；真正喂给模型的是“摘要 + 最近原文”。
