1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

## list 管理对话历史
messages = [{"role": "user", "content": "你好"},
           {"role": "assistant", "content": "你好！有什么可以帮你的吗？"}]
## dict agent状态管理
agent_state = {
    "user_id": "user_12345",
    "session_id": "sess_abc",
    "messages": [
        {"role": "user", "content": "帮我查天气"},
        {"role": "assistant", "content": "好的，请问是哪个城市？"}
    ],
    "context": {
        "intent": "weather_query",
        "location": None,
        "confidence": 0.92
    },
    "tool_calls": [],
    "metadata": {
        "start_time": "2024-10-27 10:00:00",
        "turn_count": 1
    }
}
## 元组 不可变数据
message_format = ("role", "content", "timestamp")
## set 去重和快速查找
user_interests = {"科技", "旅游", "美食", "科技", "运动"}
## 【嵌套TypedDict】LangGraph状态的标准定义方式
## 定义清晰的状态结构,便于团队协作和维护
class AgentState(TypedDict):
    messages: List[Message]  # 消息列表,每个元素都是Message类型
    intent: str
    next_node: str

1
2
3
4
5

LangGraph / LangChain   ← 核心逻辑
        ↓
      Gradio            ← Demo / 对话
        ↓
   Streamlit / FastAPI  ← 产品化

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

class ConversationState(BaseModel):
    messages: Annotated[List[str], trim_messages(20)] = []
def trim_messages(max_len: int):
    def reducer(old: List[str], new: List[str]):
        merged = old + new
        return merged[-max_len:]
    return reducer

def summarizer_node(state: ConversationState):
    if len(state.messages) < 15:
        return {}

    prompt = f"""
    当前摘要：
    {state.summary or "无"}

    请将以下对话合并成一个更精炼的摘要：
    {state.messages}
    """

    new_summary = llm(prompt)

    return {
        "summary": new_summary,
        "messages": []  # 清空窗口
    }


def semantic_filter(messages, query_embedding, k=5):
    scored = [
        (cosine_sim(m.embedding, query_embedding), m)
        for m in messages
    ]
    return [m.content for _, m in sorted(scored, reverse=True)[:k]]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

from pydantic import BaseModel, ConfigDict
from datetime import datetime
from typing import List

class BaseState(BaseModel):
    """所有 State 的统一基类"""

    model_config = ConfigDict(
        validate_assignment=True,   # 赋值即校验
        extra="forbid",              # 禁止野字段
        frozen=False                # LangGraph 需要可变
    )

    trace: List[str] = []
    updated_at: datetime = datetime.utcnow()

class InputState(BaseState):
    query: str = Field(..., description="User original input")
    user_id: str | None = None

class WorkingState(InputState):
    # Planner
    plan: Optional[str] = Field(
        None, description="High level plan generated by planner"
    )

    # Retriever
    documents: List[str] = Field(
        default_factory=list,
        description="Retrieved documents"
    )

    # Memory / Reasoning
    thoughts: List[str] = Field(
        default_factory=list,
        description="Internal reasoning trace"
    )

    # Control
    next_action: Optional[str] = None

class OutputState(WorkingState):
    answer: str = Field(..., description="Final user-facing answer")

    @model_validator(mode="after") # 在所有字段都准备好之后，对整个 State 的‘存在合法性’做最终审判。
    def check_ready(self):
        if not self.plan:
            raise ValueError("Output generated without plan")
        return self

1
2
3
4
5
6
7

from pydantic import BaseModel
from typing import List, Annotated
from langgraph.graph import add_messages

class AgentState(BaseModel):
    messages: Annotated[List[str], add_messages] = []
    current_step: str | None = None

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

## 手动方式
class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

## 快捷方式（推荐）
class State(MessagesState):
    pass  # 自动包含 messages 字段

result = add_messages(initial_messages, new_message)
delete_messages = [RemoveMessage(id=m.id) for m in messages[:-2]]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

## 在node中
graph.add_node("planner", planner_node, input_schema=PlannerState)
graph.add_node("executor", executor_node, input_schema=ExecutorState)
graph.add_node("reviewer", reviewer_node, input_schema=ReviewerState)

## 在state中
class ConversationState(BaseModel):
    messages: Annotated[List[str], add_messages] = []

class PlanningState(BaseModel):
    plan: str | None = None

class ExecutionState(BaseModel):
    observations: Annotated[List[str], add_messages] = []

class ReviewState(BaseModel):
    final_answer: str | None = None

## State Nesting（嵌套 State），本质是 组合（composition）而不是继承
class GlobalState(BaseModel): 
    conversation: ConversationState = ConversationState()
    planning: PlanningState = PlanningState()
    execution: ExecutionState = ExecutionState()
    review: ReviewState = ReviewState() 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

import sqlite3

## 创建内存数据库（程序结束后消失）
conn = sqlite3.connect(":memory:", check_same_thread=False)

## 下载示例数据库（如果不存在）
!mkdir -p state_db && [ ! -f state_db/example.db ] && wget -P state_db https://github.com/langchain-ai/langchain-academy/raw/main/module-2/state_db/example.db

## 连接到本地数据库文件
db_path = "state_db/example.db"
conn = sqlite3.connect(db_path, check_same_thread=False)

from langgraph.checkpoint.sqlite import SqliteSaver

## 创建 checkpointer
memory = SqliteSaver(conn)

CREATE TABLE checkpoints (
    thread_id TEXT,      -- 对话线程 ID
    checkpoint_id TEXT,  -- 检查点 ID
    state BLOB,          -- 序列化的状态数据
    timestamp DATETIME   -- 时间戳
);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

## interrupt_before 审批，在某个节点或工具真正执行之前，强制暂停整个 Graph 的运行。
## interrupt_after 调试，某个节点执行完成之后，立刻中断流程
## update_state 在前两者到达的中断状态下，你可以人为修改 State，然后让 Graph 继续跑。

## interrupt 负责“停”，update_state 负责“改”，invoke 负责“继续”
graph = builder.compile(
    interrupt_before=["payment_tool"],  # 在支付前暂停
    checkpointer=memory # breakpoint依赖状态持久化，会将当前状态保存倒checkpoint
)

graph.update_state(
    thread_id,
    {"approved": True}
)

graph.invoke(None, thread_id=thread_id)

1
2
3
4
5
6

from langgraph.errors import Interrupt

def node(state):
    if not state.get("approved"):
        raise Interrupt("waiting for approval")
    return state

1
2
3
4
5

for event in graph.stream(inputs):
    ...

for update in graph.stream(inputs, stream_mode="updates"):
    print(update)

1
2
3
4
5
6
7
8
9

current = graph.get_state(thread_id) # 可以查看当前value和历史history state
print(current.values) # 当前状态
for snapshot in current.history:
    print(snapshot.values, snapshot.branch_id) # 历史状态
    
graph.set_pointer(thread_id, snapshot=current.history[1]) # 回到step之后的快照
## TODO 这里可以用update state更新状态
graph.update_state(thread_id, {"count": 100, "note": "corrected after step1"}) # 修改state
graph.invoke(None, thread_id=thread_id)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

┌─────────────────────────────────────-────┐
│           Memory Agent                   │
│                                          │
│  ┌────────────────────────────────────┐  │
│  │   短期记忆 (MemorySaver)            │  │
│  │   - 对话历史                        │  │
│  │   - 当前会话状态                     │  │
│  └────────────────────────────────────┘  │
│                                          │
│  ┌────────────────────────────────────┐  │
│  │   长期记忆 (InMemoryStore)          │  │
│  │   ┌──────────────────────────────┐ │  │
│  │   │ Profile (语义记忆) - profile，collection
│  │   └──────────────────────────────┘ │  │
│  │   ┌──────────────────────────────┐ │  │
│  │   │ ToDo Collection (语义记忆)    │ │  │
│  │   └──────────────────────────────┘ │  │
│  │   ┌──────────────────────────────┐ │  │
│  │   │ Instructions (程序性记忆)     │  │  │
│  │   └──────────────────────────────┘ │  │
│  └────────────────────────────────────┘  │
└────────────────────────────────────────-─┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

## 创建内存存储实例
in_memory_store = InMemoryStore()

## put - 保存数据
def put(
    namespace: tuple,      # 命名空间（元组）
    key: str,              # 键（字符串）
    value: dict            # 值（字典）
) -> None:
    
user_id = "1"
namespace_for_memory = (user_id, "memories")

key = str(uuid.uuid4())
value = {"food_preference": "I like pizza"}

in_memory_store.put(namespace_for_memory, key, value)

## search - 获取用户所有记忆
memories = in_memory_store.search(namespace_for_memory)
print(memories[0].dict())

## get - 获取特定的一条记忆
memory = in_memory_store.get(namespace_for_memory, key)
print(memory.dict())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

┌─────────────────────────────────────────────┐
│           Trustcall Workflow                │
│                                             │
│  1. 接收输入                                │
│     - 新的对话消息                          │
│     - 现有的 Schema（如果有）               │
│                                             │
│  2. 分析变化                                │
│     - 识别新信息                            │
│     - 识别需要更新的字段                    │
│                                             │
│  3. 生成 JSON Patch                         │
│     - 创建精确的更新操作                    │
│     - 只修改变化的部分                      │
│                                             │
│  4. 应用 Patch                              │
│     - 更新现有 Schema                       │
│     - 保留未变化的信息                      │
│                                             │
│  5. 验证结果                                │
│     - 确保符合 Schema 定义                  │
│     - 如果失败，自动修正                    │
└─────────────────────────────────────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

## 定义shcema
class UserProfile(BaseModel):
    name: Optional[str]
    interests: List[str] = Field(default_factory=list)
    expertise_level: Optional[str]

## 创建extractor，输出一定符合Pydantic/JSON Schema
profile_extractor = create_extractor(
    UserProfile,
    description="""
    Extract stable user profile information from conversation.

    Rules:
    - Only extract explicit facts
    - Do NOT guess or infer
    - Ignore temporary tasks or moods
    """
)

## 在agent中调用
def extract_profile(llm, user_input: str, existing_profile: UserProfile):
    profile = profile_extractor.invoke(
        llm,
        user_input,
        existing=existing_profile
    )
    return profile

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

Profile 模式（5.3）              Collection 模式（5.4）
    ↓                               ↓
┌────────────────┐            ┌────────────────┐
│  用户资料      │            │  记忆 1        │
│  ┌──────────┐  │            │  content: ...  │
│  │ name     │  │            └────────────────┘
│  │ location │  │            ┌────────────────┐
│  │ interests│  │            │  记忆 2        │
│  └──────────┘  │            │  content: ...  │
│  单一对象      │            └────────────────┘
└────────────────┘            ┌────────────────┐
                              │  记忆 3        │
更新 = 修改字段                │  content: ...  │
                              └────────────────┘
                              多个独立对象

                              更新 = 修改现有项
                              插入 = 添加新项

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from pydantic import BaseModel, Field

class Memory(BaseModel):
    content: str
    category: Optional[str] = None      # 可选的类别
    importance: Optional[int] = None    # 可选的重要性评分
    timestamp: datetime = Field(default_factory=datetime.now)
    
class MemoryCollection(BaseModel):
    memories: list[Memory] = Field(
        description="A list of memories about the user."
    )

model = ChatOpenAI(model="gpt-5-nano", temperature=0)
model_with_structure = model.with_structured_output(MemoryCollection)
memory_collection = model_with_structure.invoke([
    HumanMessage("My name is Lance. I like to bike.")
])    
for mem in memory_collection.memories:
    print(mem.model_dump())

## 保存到store
key = str(uuid.uuid4())  # 生成唯一 ID
value = memory_collection.memories[0].model_dump()
in_memory_store.put(namespace_for_memory, key, value)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict

## 父 Graph 的 State
class ParentState(TypedDict):
    inputs: list
    results: list

## 定义 Map 子图：对每个元素做操作
def map_task(item: int):
    return item * 2  # 简单倍数处理

## 定义 Reduce 节点
def reduce_task(state: ParentState, map_outputs):
    # 汇总所有 Map 输出
    state["results"] = map_outputs
    return state

## 构建父 Graph
builder = StateGraph(ParentState)

## Map-Reduce 节点
builder.add_map_reduce(
    name="map_reduce_example",
    items_key="inputs",          # 输入列表
    map_func=map_task,           # Map
    reduce_func=reduce_task      # Reduce
)

builder.set_entry_point("map_reduce_example")
graph = builder.compile(checkpointer=MemorySaver())

## 执行
res = graph.invoke({"inputs": [1,2,3,4], "results": []})
print(res)  # {'inputs': [1,2,3,4], 'results': [2,4,6,8]}

1
2
3
4
5

graph.add_conditional_edges(
    "source_node",           # 源节点
    condition_function,      # 返回 Send 列表的函数
    ["target_node"]         # 目标节点（Send 指向的节点）
)

1

display(Image(graph.get_graph().draw_mermaid_png()))