1
# ========== 本篇所有答案共用的开头（模型创建） ==========
2
import os
3

4
from dotenv import load_dotenv
5
from langchain.agents import create_agent
6
from langchain.chat_models import init_chat_model
7

8
load_dotenv(override=True)
9

10
model = init_chat_model(
11
    model="deepseek-v4-flash",
12
    model_provider="openai",
13
    api_key=os.getenv("DEEPSEEK_API_KEY"),
14
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
15
)
16

17
# ---------- 2-1 让 Agent 记住你 ----------
18
from langchain_core.messages import HumanMessage
19
from langgraph.checkpoint.memory import InMemorySaver
20

21
agent = create_agent(model=model, checkpointer=InMemorySaver())
22

23
same = {"configurable": {"thread_id": "1"}}          # 同一个会话 ID
24
r1 = agent.invoke({"messages": [HumanMessage("你好，我叫小明")]}, config=same)
25
print(f"[同会话] 第 1 轮后 state 消息数: {len(r1['messages'])}")
26
r2 = agent.invoke({"messages": [HumanMessage("我叫什么？")]}, config=same)
27
print("[同会话] 回答:", r2["messages"][-1].content)
28
print(f"[同会话] 第 2 轮后 state 消息数: {len(r2['messages'])}")   # 历史被自动带上了
29

30
other = {"configurable": {"thread_id": "2"}}         # 换一个会话 ID
31
r3 = agent.invoke({"messages": [HumanMessage("我叫什么？")]}, config=other)
32
print("[新会话] 回答:", r3["messages"][-1].content)
33
print(f"[新会话] state 消息数: {len(r3['messages'])}")            # 只有本轮，新会话是白纸
34

35
# ---------- 2-2 裁剪：首条 + 最近几条（保证人机成对） ----------
36
from typing import Any
37

38
from langchain.agents import AgentState
39
from langchain.agents.middleware import before_model
40
from langchain_core.messages import RemoveMessage
41
from langgraph.graph.message import REMOVE_ALL_MESSAGES
42
from langgraph.runtime import Runtime
43

44
@before_model
45
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
46
    messages = state["messages"]
47
    if len(messages) <= 4:
48
        return None
49

50
    first_msg = messages[0]
51
    # 让片段以 AI 消息开头：偶数取 3 条（[A,H,A]）、奇数取 4 条（[A,H,A,H]）
52
    recent_messages = messages[-3:] if len(messages) % 2 == 0 else messages[-4:]
53
    print(f"   [裁剪] {len(messages)} 条 → 首条 + 最近 {len(recent_messages)} 条")
54

55
    return {
56
        "messages": [
57
            RemoveMessage(id=REMOVE_ALL_MESSAGES),   # 先全部清掉
58
            first_msg,
59
            *recent_messages,                        # 再放回要保留的
60
        ]
61
    }
62

63
agent_trim = create_agent(model=model, middleware=[trim_messages], checkpointer=InMemorySaver())
64
cfg_trim = {"configurable": {"thread_id": "trim"}}
65
for i in range(1, 6):
66
    rt = agent_trim.invoke({"messages": [HumanMessage(f"第 {i} 句话")]}, config=cfg_trim)
67
    print(f"第 {i} 轮后 state 消息数: {len(rt['messages'])}")
68
# 涨到 6 条之后被按住，稳定在同一水平，不再无限增长
69

70
# ---------- 2-3 删除：永久遗忘 ----------
71
from langchain.agents.middleware import after_model
72

73
@after_model
74
def delete_old_messages(state: AgentState, runtime: Runtime) -> dict | None:
75
    messages = state["messages"]
76
    if len(messages) > 5:
77
        to_delete = len(messages) - 5
78
        print(f"   [删除] 永久删除最早的 {to_delete} 条")
79
        # 逐条标记删除，返回出去让框架真正移除
80
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:to_delete]]}
81
    return None
82

83
agent_del = create_agent(model=model, middleware=[delete_old_messages], checkpointer=InMemorySaver())
84
cfg_del = {"configurable": {"thread_id": "delete"}}
85
for i in range(1, 4):
86
    rd = agent_del.invoke({"messages": [HumanMessage(f"第 {i} 句话")]}, config=cfg_del)
87
    left = [m.content for m in rd["messages"] if m.type == "human"]
88
    print(f"第 {i} 轮后 state 消息数: {len(rd['messages'])}，剩下的用户消息 {left}")
89
# 第 3 轮触发删除：最早那条真的从 state 里消失了（下一轮也再也看不到）
90

91
# ---------- 2-4 亲眼看看 state 里那两个字段 ----------
92
from pydantic import BaseModel, Field
93

94
from langchain.agents.middleware import after_agent, wrap_tool_call
95
from langchain_core.messages import AIMessage
96
from langchain.tools import tool
97
from langchain.tools.tool_node import ToolCallRequest
98

99
class WeatherInfo(BaseModel):
100
    """城市天气情况"""
101

102
    city: str = Field(description="城市名称")
103
    temperature: str = Field(description="气温")
104
    desc: str = Field(description="当日天气概述")
105

106
@tool(parse_docstring=True)
107
def get_weather(city: str):
108
    """获取当日天气
109

110
    Args:
111
        city: 城市名称
112
    """
113
    return f"[{city}] 今天气温9~16度，万里无云，天气不错适合外出"
114

115
@before_model(can_jump_to=["tools"])                 # ← 要跳转就得先声明能跳到哪
116
def direct_tool_call(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
117
    last_msg = state["messages"][-1]
118
    if isinstance(last_msg, HumanMessage) and "天气" in last_msg.text and "北京" in last_msg.text:
119
        fake_tool_call = AIMessage(                  # 人工伪造一条"模型想调工具"的消息
120
            content="人工构造的消息",
121
            tool_calls=[{"name": "get_weather", "args": {"city": "北京"}, "id": "direct_call_id"}],
122
        )
123
        return {"messages": [fake_tool_call], "jump_to": "tools"}    # ← 跳过模型，直奔工具节点
124
    return None
125

126
@wrap_tool_call
127
def check_before_tool(request: ToolCallRequest, handler):
128
    print("--- 工具调用前 ---")
129
    print(f"{request.state.get('jump_to', None) = }")            # 此刻跳转意图还在
130
    print(f"{request.state.get('structured_response', None) = }")
131
    return handler(request)
132

133
@after_agent
134
def check_after_agent(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
135
    print("--- Agent 跑完之后 ---")
136
    for msg in state["messages"]:
137
        print(f"   {type(msg).__name__:15s} {str(msg.content)[:50]}")
138
    print(f"{state.get('jump_to', None) = }")                    # 跳完就清掉了 → None
139
    print(f"{state.get('structured_response', None) = }")        # 跑完才写进去 → 非空
140
    return None
141

142
agent_state = create_agent(
143
    model=model,
144
    response_format=WeatherInfo,          # 开启结构化输出，结果写进 structured_response
145
    middleware=[direct_tool_call, check_before_tool, check_after_agent],
146
    tools=[get_weather],
147
)
148
agent_state.invoke({"messages": [HumanMessage("请帮我查询北京当日天气")]})

1
# ========== 综合题答案（模型创建部分同上） ==========
2
import os
3
from typing import Any
4

5
from dotenv import load_dotenv
6
from langchain.agents import AgentState, create_agent
7
from langchain.agents.middleware import before_model
8
from langchain.chat_models import init_chat_model
9
from langchain_core.messages import HumanMessage, RemoveMessage
10
from langgraph.checkpoint.memory import InMemorySaver
11
from langgraph.runtime import Runtime
12

13
load_dotenv(override=True)
14

15
model = init_chat_model(
16
    model="deepseek-v4-flash",
17
    model_provider="openai",
18
    api_key=os.getenv("DEEPSEEK_API_KEY"),
19
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
20
)
21

22
# ---------- 3-1 客服助手：给上下文管理加一条"白名单" ----------
23
KEEP_KEYWORD = "订单号"
24

25
@before_model
26
def filter_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
27
    """自定义过滤：首条 + 命中关键词的消息永久保留，其余只留最近 4 条"""
28
    messages = state["messages"]
29
    if len(messages) <= 6:
30
        return None
31

32
    # 注意：system_prompt= 传进去的系统提示并不在 state["messages"] 里，
33
    # 所以这里的 messages[0] 就是最早的那条用户消息
34
    keep_ids = {messages[0].id}                                              # ① 最早的一条
35
    keep_ids |= {m.id for m in messages if KEEP_KEYWORD in str(m.content)}    # ② 关键事实白名单
36
    keep_ids |= {m.id for m in messages[-4:]}                                # ③ 最近 4 条
37

38
    dropped = [m for m in messages if m.id not in keep_ids]
39
    if not dropped:
40
        return None
41
    print(f"   [过滤] {len(messages)} 条 → 保留 {len(keep_ids)} 条、丢掉 {len(dropped)} 条")
42

43
    # 没进白名单的消息——永久删掉（这里没有内置保护，规则全靠自己写）
44
    return {"messages": [RemoveMessage(id=m.id) for m in dropped]}
45

46
SYSTEM = "你是客服助手，记住用户告诉你的订单号等关键信息，并在被问到时准确回答。"
47

48
agent_cs = create_agent(
49
    model=model,
50
    system_prompt=SYSTEM,
51
    middleware=[filter_messages],
52
    checkpointer=InMemorySaver(),
53
)
54
cfg_cs = {"configurable": {"thread_id": "cs-1"}}
55

56
print("[1] 先打个招呼")
57
agent_cs.invoke({"messages": [HumanMessage("你好，在吗？")]}, config=cfg_cs)
58
print("[2] 报订单号")
59
print("  AI:", agent_cs.invoke({"messages": [HumanMessage("我的订单号是 A12345")]}, config=cfg_cs)["messages"][-1].content[:60])
60
print("[3] 立刻追问（记得）")
61
print("  AI:", agent_cs.invoke({"messages": [HumanMessage("我的订单号是多少？")]}, config=cfg_cs)["messages"][-1].content[:60])
62
print("[4] 再聊 5 轮无关内容")
63
for q in ["今天天气怎么样？", "你们发货快吗？", "支持退货吗？", "有优惠活动吗？", "客服几点下班？"]:
64
    agent_cs.invoke({"messages": [HumanMessage(q)]}, config=cfg_cs)
65
print("[5] 最后再问一次订单号")
66
print("  AI:", agent_cs.invoke({"messages": [HumanMessage("我的订单号是多少？")]}, config=cfg_cs)["messages"][-1].content[:80])
67
# 白名单规则生效 → 订单号还在；把 KEEP_KEYWORD 那条规则去掉再跑一遍，[5] 就答不上来了
68
# ——这就是"关键事实该进长期记忆"的原因（下一篇）