1
from langgraph.checkpoint.memory import InMemorySaver
2
from langchain.agents import create_agent
3
from langchain_core.messages import HumanMessage
4

5
checkpointer = InMemorySaver()
6

7
agent = create_agent(
8
    model=model,
9
    checkpointer=checkpointer,       # ← 1. 加检查点
10
)
11

12
config = {"configurable": {"thread_id": "1"}}    # ← 2. 指定会话 ID
13

14
agent.invoke({"messages": [HumanMessage("你好，我叫小明")]}, config=config)
15
agent.invoke({"messages": [HumanMessage("我叫什么？")]}, config=config)   # ← 3. 用同一个 config
16
# 这次它就能答上来了

1
from typing import Any
2

3
from langchain.agents import AgentState, create_agent
4
from langchain.agents.middleware import before_model
5
from langchain_core.messages import RemoveMessage
6
from langgraph.graph.message import REMOVE_ALL_MESSAGES
7
from langgraph.runtime import Runtime
8

9
@before_model
10
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
11
    messages = state["messages"]
12
    if len(messages) <= 3:
13
        return None
14

15
    first_msg = messages[0]                       # 保留开头（比如系统提示）
16
    recent_messages = messages[-3:]                # 保留最近几条
17
    new_messages = [first_msg, *recent_messages]
18

19
    return {
20
        "messages": [
21
            RemoveMessage(id=REMOVE_ALL_MESSAGES),  # 先全部清掉
22
            *new_messages,                          # 再放回要保留的
23
        ]
24
    }
25

26
agent = create_agent(model=model, middleware=[trim_messages], checkpointer=InMemorySaver())

1
from typing import Any, Callable
2

3
from pydantic import BaseModel, Field
4

5
from langchain.agents import AgentState, create_agent
6
from langchain.agents.middleware import after_agent, before_model, wrap_tool_call
7
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
8
from langchain.tools import tool
9
from langchain.tools.tool_node import ToolCallRequest
10
from langgraph.runtime import Runtime
11
from langgraph.types import Command
12

13
class WeatherInfo(BaseModel):
14
    """城市天气情况"""
15
    city: str = Field(description="城市名称")
16
    temperature: str = Field(description="气温")
17
    desc: str = Field(description="当日天气概述")
18

19
@tool(parse_docstring=True)
20
def get_weather(city: str):
21
    """获取当日天气
22

23
    Args:
24
        city: 城市名称
25
    """
26
    return f"[{city}] 今天气温9~16度，万里无云，天气不错适合外出"
27

28
@before_model(can_jump_to=["tools"])
29
def direct_tool_call(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
30
    last_msg = state["messages"][-1]
31
    if isinstance(last_msg, HumanMessage) and "天气" in last_msg.text and "北京" in last_msg.text:
32
        fake_tool_call = AIMessage(                 # 人工伪造一条"模型想调工具"的消息
33
            content="人工构造的消息",
34
            tool_calls=[{"name": "get_weather", "args": {"city": "北京"}, "id": "direct_call_id"}],
35
        )
36
        return {"messages": [fake_tool_call], "jump_to": "tools"}    # ← 直接跳工具节点
37
    return None
38

39
@wrap_tool_call
40
def first_check(
41
    request: ToolCallRequest,
42
    handler: Callable[[ToolCallRequest], ToolMessage | Command],
43
) -> ToolMessage | Command:
44
    print("=" * 30, '-> In first_check Middleware <-', "=" * 30)
45
    print(f"{request.state.get('jump_to', None) = }")
46
    print(f"{request.state.get('structured_response', None) = }")
47
    return handler(request)
48

49
@after_agent
50
def final_check(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
51
    print("=" * 30, '-> In final_check Middleware <-', "=" * 30)
52
    for msg in state["messages"]:
53
        msg.pretty_print()
54
    print(f"{state.get('jump_to', None) = }")
55
    print(f"{state.get('structured_response', None) = }")
56
    return None
57

58
agent = create_agent(
59
    model=model,
60
    response_format=WeatherInfo,        # ← 开启结构化输出
61
    middleware=[direct_tool_call, first_check, final_check],
62
    tools=[get_weather],
63
)
64

65
agent.invoke({"messages": [HumanMessage("请帮我查询北京当日天气")]})

1
============================== -> In first_check Middleware <- ==============================
2
request.state.get('jump_to', None) = 'tools'
3
request.state.get('structured_response', None) = None
4
============================== -> In final_check Middleware <- ==============================
5
================================ Human Message =================================
6
请帮我查询北京当日天气
7
================================== Ai Message ==================================
8
人工构造的消息
9
Tool Calls:
10
  get_weather (direct_call_id)
11
 Call ID: direct_call_id
12
  Args:
13
    city: 北京
14
================================= Tool Message =================================
15
Name: get_weather
16
[北京] 今天气温9~16度，万里无云，天气不错适合外出
17
================================== Ai Message ==================================
18
{"city":"北京","temperature":"9~16℃","desc":"万里无云，天气不错适合外出"}
19
============================== -> 消息打印完毕 <- ==============================
20
state.get('jump_to', None) = None
21
state.get('structured_response', None) = WeatherInfo(city='北京', temperature='9~16℃', desc='万里无云，天气不错适合外出')