W5-D2 · 条件边：让图自己选路

1
2
3
4
5
6

  D1 的图（传送带）        D2 的图（分拣机器人）
  ───────────────         ───────────────────
  START → A → B → END     START →┬→ A → END
                                  ├→ B → END
                                  └→ C → END
                          看一眼 state，决定扔哪条道

1

builder.add_edge("upper", "join")     # upper 跑完必去 join

1
2
3
4
5

def hub(state):
    q = state["question"]
    if any(k in q for k in ["+", "搜索"]):
        return {"answer": handle_math_or_tool(state)}   # 在一个节点里全干了
    return {"answer": handle_chat(state)}

1
2

  塞 if：   决策逻辑在节点内部 → 看不见、改起来要动业务代码
  条件边：  决策逻辑在"边"上 → 图自己决定走哪条 → 拓扑和业务解耦

1
2
3
4
5

builder.add_conditional_edges(
    "source_node",   # ① 从哪个节点出来（岔路口）
    route_decision,  # ② 路由函数：读 state，返回字符串
    {"math": "math_node", "chat": "chat_node"},  # ③ path_map：返回值 → 真实节点名
)

1
2
3

  你声明时：  add_node("math_node", math_node)     ← 注册"名字→函数"
  运行时：    router(state) → "math_node"          ← 只报名字
  图引擎：    "math_node" → 找到函数 → 执行         ← 引擎负责

1
2
3
4
5
6
7
8

# 情况 A：返回代号，必须写 path_map
def router(state): return "math"           # "math" 不是节点名
add_conditional_edges("classify", router,
                      {"math": "math_node", "chat": "chat_node"})   # ✅ 必须

# 情况 B：返回值就是节点名，可省
def router(state): return "math_node"      # 正好是 add_node 用的名字
add_conditional_edges(START, router)       # ✅ 省略

 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

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class QState(TypedDict):
    question: str
    answer: str

def classify_router(state: QState) -> str:
    """路由函数：只读 state，返回字符串（此处省略 path_map，因为返回值 = 节点名）"""
    q = state["question"].lower()
    if any(k in q for k in ["+", "-", "*", "/", "等于", "几"]):
        return "math_node"
    if any(k in q for k in ["搜索", "什么是"]):
        return "tool_node"
    return "chat_node"

def math_node(state: QState) -> dict:
    return {"answer": "[数学] " + state["question"]}

def tool_node(state: QState) -> dict:
    return {"answer": "[工具] " + state["question"]}

def chat_node(state: QState) -> dict:
    return {"answer": "[闲聊] " + state["question"]}

b = StateGraph(QState)
for name, fn in [("math_node", math_node), ("tool_node", tool_node), ("chat_node", chat_node)]:
    b.add_node(name, fn)

# START 直接挂条件边（入口路由）：省掉"先到分类节点再路由"那步
b.add_conditional_edges(START, classify_router)
for name in ["math_node", "tool_node", "chat_node"]:
    b.add_edge(name, END)

g = b.compile()

for q in ["1+1 等于几", "搜索 LangGraph", "今天天气不错"]:
    print(g.invoke({"question": q}))

1
2
3

{'question': '1+1 等于几',    'answer': '[数学] 1+1 等于几'}
{'question': '搜索 LangGraph', 'answer': '[工具] 搜索 LangGraph'}
{'question': '今天天气不错',   'answer': '[闲聊] 今天天气不错'}

1
2
3

for q in ["1+1 等于几", "搜索 LangGraph", "今天天气不错"]:
    for step in g.stream({"question": q}, stream_mode="updates"):
        print(step)

1
2
3

{'math_node': {'answer': '[数学] 1+1 等于几'}}
{'tool_node': {'answer': '[工具] 搜索 LangGraph'}}
{'chat_node': {'answer': '[闲聊] 今天天气不错'}}

1
2

  ❌ 不是：三个节点都跑，最后挑一个结果
  ✅ 是：   只跑被选中的那一个

1
2
3
4
5

  START → classify → ┬→ math_node  → END
                     ├→ tool_node  → END
                     └→ chat_node  → END
      ↑ 先跑个 classify 节点，把 "math" 写进 state["qtype"]
          再由 router 读 qtype 决定去哪

1
2
3
4

  START →┬→ math_node  → END
         ├→ tool_node  → END
         └→ chat_node  → END
      ↑ 分类逻辑直接写进 router，START 就分叉

1
2
3
4
5
6
7
8
9

  W3 的 MemoryManager（命令式）：          画成图（声明式）：
  def get_context(...):                   START
      if len > 20: 压缩                        │
      elif "记住":  存长期              decide_path(state)
      else:        短期                        │
                                  ┌───────────┼───────────┐
                                  │           │           │
                                  ▼           ▼           ▼
                               compress     long        short

 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
50
51
52
53
54
55

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class MemoryState(TypedDict):
    messages: list[dict]
    compressed_summary: str
    long_term_hit: str
    decision: str

def decide_path(state: MemoryState):
    """路由函数：读 state，返回节点名（省略 path_map 因为返回值 = 节点名）"""
    if len(state["messages"]) > 20:
        return "compress"                                  # ① 消息太多 → 压缩
    recent = [m.get("content", "") for m in state["messages"]]
    if any("记住" in c for c in recent):
        return "long"                                      # ② 最近有"记住" → 存长期
    return "short"                                         # ③ 默认 → 短期窗口

def short_node(state: MemoryState) -> dict:
    return {"decision": "short"}                           # 只写自己那格，别的字段保留

def long_node(state: MemoryState) -> dict:
    return {"decision": "long", "long_term_hit": "[模拟] 从向量库召回用户画像"}

def compress_node(state: MemoryState) -> dict:
    return {"decision": "compress", "compressed_summary": "[模拟] 老消息已摘要"}

b = StateGraph(MemoryState)
for name, fn in [("short", short_node), ("long", long_node), ("compress", compress_node)]:
    b.add_node(name, fn)

b.add_conditional_edges(START, decide_path)
for name in ["short", "long", "compress"]:
    b.add_edge(name, END)

g = b.compile()

# 三条测试路（messages 统一用 dict，别混 str）
tests = {
    "走短期": {"messages": [{"role": "user", "content": "今天天气不错"}],
               "compressed_summary": "", "long_term_hit": "", "decision": ""},
    "走长期": {"messages": [{"role": "user", "content": "记住我住北京"},
                            {"role": "assistant", "content": "好的"}],
               "compressed_summary": "", "long_term_hit": "", "decision": ""},
    "走压缩": {"messages": [{"role": "user", "content": f"普通消息{i}"} for i in range(21)],
               "compressed_summary": "", "long_term_hit": "", "decision": ""},
}

for label, data in tests.items():
    print(f"\n=== {label} ===")
    for s in g.stream(data, stream_mode="updates"):   # 看实际触发了哪个节点
        print("  ", s)
    out = g.invoke(data)
    extra = out.get("long_term_hit") or out.get("compressed_summary") or ""
    print(f"  decision={out.get('decision')}  {extra}")

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

                    START
                      │
                      ▼
              decide_path(state)
                      │
        ┌─────────────┼─────────────┐
        │             │             │
  len(messages)>20  含"记住"      其他
        │             │             │
        ▼             ▼             ▼
   compress          long          short
   decision=         decision=     decision=
   "compress"        "long"        "short"
   +摘要             +召回结果     （只写 decision）
        │             │             │
        └─────────────┼─────────────┘
                      ▼
                     END

1
2
3
4
5
6

if len(state["messages"]) > 20:      # 先看长度
    return "compress"
recent = ...
if any("记住" in c for c in recent):  # 再看内容
    return "long"
return "short"                        # 兜底

1
2
3

def short_node(state):    return {"decision": "short"}
def long_node(state):     return {"decision": "long", "long_term_hit": "..."}
def compress_node(state): return {"decision": "compress", "compressed_summary": "..."}

1
2
3

  走 short   → 最终 state 里 long_term_hit、compressed_summary 保持 ""
  走 long    → long_term_hit 被填上，compressed_summary 还是 ""
  走 compress→ compressed_summary 被填上，long_term_hit 还是 ""

1
2
3
4
5

def long_node(state: MemoryState) -> dict:
    # 模拟 → 真实：调 W3 的长期记忆写入
    return {"decision": "long", "long_term_hit": "[模拟] 从向量库召回用户画像"}
    #                      ↓
    # return {"decision": "long", "long_term_hit": str(mm.long_term.recall(...))}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

class MemoryState(TypedDict):
    messages: list[dict]
    compress_summary: str        # ← 声明：compress_summary（无 ed）
    long_term_hit: str
    decision: str

tests = {
    "走短期": {"messages": [...],
               "compressed_summary": "",    # ← 传入：compressed_summary（有 ed）
               ...}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# 方案 A：统一成 compressed_summary（推荐）
class MemoryState(TypedDict):
    messages: list[dict]
    compressed_summary: str
    long_term_hit: str
    decision: str

def compress_node(state: MemoryState) -> dict:
    return {"decision": "compress", "compressed_summary": "[模拟] 老消息已摘要"}

# 方案 B：统一成 compress_summary
#   → 那 tests 里也要改成 "compress_summary": ""

1
2
3

# ❌ 默认值给成空字典
recent = [m.get("content", {}) for m in state["messages"]]
if any("记住" in c for c in recent):

1
2

# ✅ 默认值用空字符串
recent = [m.get("content", "") for m in state["messages"]]

1
2
3
4
5
6

# ❌ 混用
{"messages": [{"content": "记住我住北京"}, "x"*30]}
#                                          ↑ 这个元素没有 .get()

# ✅ 统一 dict
{"messages": [{"role": "user", "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
26
27
28
29
30
31
32

# ===== 条件边三要素 =====
builder.add_conditional_edges(
    "source_node",        # ① 从哪个节点出来
    route_fn,             # ② 路由函数 (state) -> str，只读
    {"math": "math_node"} # ③ path_map（返回值=节点名时可省）
)

# ===== 入口条件路由：START 直接分叉（省掉分类节点）=====
b.add_conditional_edges(START, classify_router)
# router 返回 "math_node" 这种节点名 → 可省 path_map

# ===== 每个分支都要连 END =====
for name in ["math_node", "tool_node", "chat_node"]:
    b.add_edge(name, END)

# ===== 看路由走了哪条路 =====
for step in g.stream(data, stream_mode="updates"):
    print(step)     # {'math_node': {...}}   ← 只有被选中的节点出现

# ===== stream_mode 对比 =====
# "updates"（默认）: {节点名: 该节点更新的字段}      ← 看"谁改了什么"
# "values"        : 每个节点后的完整 state          ← 看"工单长什么样"

# ===== 路由函数铁律 =====
# 1. 返回 str（或 list[str]），不是函数对象
# 2. 只读 state，不改 state
# 3. 返回值 = 节点名 → 可省 path_map；是代号 → 必须写

# ===== MemoryState 雏形（记忆画成图）=====
# 路由 decide_path：消息>20 → compress；含"记住" → long；否则 → short
# 三个节点各写一格：short 只写 decision，long 加 long_term_hit，
#                  compress 加 compressed_summary —— 未返回的字段原样保留

1
2
3
4
5
6

  今天：  START → decide_path →┬→ short     （雏形：节点只写 decision）
                               ├→ long
                               └→ compress

  下一步：把 W3 的真实逻辑填进节点 → MemoryManager 成图
  D3：    把 Function Calling 接进图 → 图开始会"干活"