W5-D1 · 从"写流程"到"声明流程"

1
2

while ai.tool_calls:          # 模型还要调工具？
    执行工具 → 回填 ToolMessage → 再问模型

1
2
3
4
5

  while ai.tool_calls:
      for call in ai.tool_calls:     ← 第 1 步在这行
          result = ...
          messages.append(...)       ← 第 2 步在这行
      ai = model.invoke(messages)    ← 第 3 步在这行

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

┌──────────────────────────────────────────────────────┐
│  LCEL（W4）                                            │
│  prompt | model | parser                              │
│  · 无环：数据流单向，走完就结束                         │
│  · 无状态：每次 invoke 都是干净调用                     │
│  · 擅长：单个步骤内部的"加工流水线"                    │
├──────────────────────────────────────────────────────┤
│  LangGraph（W5）                                       │
│  StateGraph：node + edge + state                      │
│  · 有环：能绕回去重复执行                              │
│  · 有状态：State 在节点间传递并被更新                   │
│  · 擅长：多个步骤之间的"编排调度"                      │
└──────────────────────────────────────────────────────┘
           ↓ 不是替代，是分工 ↓
  步骤内部用 LCEL，步骤之间用 LangGraph

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

  投料口              工位 A            工位 B            成品出口
  START    ──▶    [ think ]   ──▶   [ respond ]   ──▶    END
                       │                  │
                       ▼                  ▼
                  填"answer"格       在 answer 后面追加

  工单（State）流转：
  {question: "什么是 LangGraph?"}                    ← 投料时
  {question: "...", answer: "[已思考] ..."}           ← 过完 think
  {question: "...", answer: "[已思考] ... → 回复完成"} ← 过完 respond → 出库

 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

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

# ── State：定义"工单上有哪些格子" ─────────────────────────
class State(TypedDict):
    question: str      # 入口时填好（invoke 传入）
    answer: str        # 由节点往这里填

# ── node：普通函数，输入整张工单，返回"我要更新的格子" ────
def think(state: State) -> dict:
    return {"answer": f"[已思考] {state['question']}"}   # ⚠️ 只返回 dict

def respond(state: State) -> dict:
    return {"answer": state["answer"] + " → 回复完成"}

# ── 搭图：加工序 → 连传送带 ───────────────────────────────
builder = StateGraph(State)
builder.add_node("think", think)          # 给工序起名字 + 绑函数
builder.add_node("respond", respond)
builder.add_edge(START, "think")          # 投料口 → 第一道工序
builder.add_edge("think", "respond")      # think 干完 → respond
builder.add_edge("respond", END)          # respond 干完 → 出口

# ── compile：图纸变生产线（图因此变成了一个 Runnable）─────
graph = builder.compile()

# ── invoke：放一张工单，跑到底 ─────────────────────────────
out = graph.invoke({"question": "什么是 LangGraph?"})
print(out)   # {'question': '...', 'answer': '[已思考] ... → 回复完成'}

1
2

def think(state: State) -> dict:
    return {"answer": f"[已思考] {state['question']}"}

1
2

  ❌ 误解：return 的是"新的完整 state"
  ✅ 正解：return 的是"我要更新 state 的哪些格子"（partial update）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# ❌ 错误：返回整个 state（而且没带的字段会丢失语义）
def think_bad(state: State) -> State:
    return {"question": state["question"], "answer": "..."}   # 能跑但画蛇添足

# ✅ 正确：只返回要更新的字段
def think(state: State) -> dict:
    return {"answer": "..."}

# ✅ 也合法：什么都不更新
def noop(state: State) -> dict:
    return {}

1
2
3

builder.add_edge(START, "think")
builder.add_edge("think", "respond")
builder.add_edge("respond", END)

1
2
3

  START ──▶ think ──▶ respond ──▶ END
   ↑                              ↑
  投料口（固定）                  出口（固定）

1

graph = builder.compile()

1
2
3
4
5

# 在 lg_basics.py 末尾追加
print("\n--- 逐节点观察流转 ---")
for chunk in graph.stream({"question": "你好"}):
    for node_name, update in chunk.items():   # chunk = {节点名: 它更新了什么}
        print(f"[{node_name}] 填了 → {update}")

1
2
3

--- 逐节点观察流转 ---
[think] 填了 → {'answer': '[已思考] 你好'}
[respond] 填了 → {'answer': '[已思考] 你好 → 回复完成'}

1

print(graph.get_graph().draw_ascii())

1

START → think → respond → END

1
2
3
4
5
6
7
8
9

# ❌ 手写循环思维的残留
answer = ""                      # 全局变量存答案

def think(state: State) -> dict:
    global answer                # 改全局变量！
    answer = f"[已思考] {state['question']}"
    return {}                    # 什么都不返回

# 结果：state["answer"] 永远是空的

1
2
3
4
5
6
7

  正确心智：
  ┌─────────────┐
  │    node     │
  │  读 state   │ ← 入参（只读，别改）
  │  算一算     │
  │  return {}  │ ← 唯一出口（要更新的字段）
  └─────────────┘

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

  方向一：LangChain → LangGraph
  ┌──────────────────────────────────────┐
  │  把 W4 的 LCEL 链塞进图里当节点       │
  │  chain = prompt | llm | parser        │
  │  node 函数里 chain.invoke(...)        │
  └──────────────────────────────────────┘

  方向二：LangGraph → LangChain
  ┌──────────────────────────────────────┐
  │  把一张图塞进 LCEL 链里当一环         │
  │  big_chain = prompt | graph | parser  │
  └──────────────────────────────────────┘

1

print(graph.get_graph().draw_ascii())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

class State(TypedDict):
    question: str
    answer: str
    turns: int          # 新增

def think(state: State) -> dict:
    return {
        "answer": f"[已思考] {state['question']}",
        "turns": state["turns"] + 1,     # 读旧值，写新值
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

  invoke 传入          {question: "..."} , turns: 0
       │
       ▼
  think 节点读到        state["turns"] == 0     ← 拿到的是上一手传下来的值
       │
       │  return {"turns": 0 + 1}
       ▼
  工单变成             turns: 1                ← 在同一张单子上改
       │
       ▼
  下一个节点读到        state["turns"] == 1     ← 能看见 think 的修改

 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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

# W5/lg_c1_w4_chain.py —— W5-D1 步骤 C：把 W4 的链包成 LangGraph 节点
# 一句话：验证 LangChain ↔ LangGraph 互通 —— 链能进图，图能跑链，记忆照常工作
import os, sys
from pathlib import Path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))  # 回项目根，同 W4
from dotenv import load_dotenv
load_dotenv(Path(__file__).parent.parent / ".env", override=True)

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from memory.memory_manager import MemoryManager      # W3 组件，原样复用

# ══════════ 1. W4 组件照搬（一行不改） ══════════
mm = MemoryManager(window=8, max_tokens=1500, keep=4)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是有记忆，能调工具的助手。历史对话和检索到用户事实、喜好和用户说记下的内容都在history里"),
    ("placeholder", "{history}"),   # 历史消息原地展开
    ("human", "{question}"),
])

# ⚠️ 关键：这里用【不绑工具】的 llm，不是 W4 那个 bind_tools 的 model！
llm = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    temperature=0,
)

# W4 的桥：dict 消息 → BaseMessage（原样搬）
def _to_lc(m: dict):
    role, content = m.get("role", "user"), m.get("content", "")
    if role == "system":
        return SystemMessage(content=content)
    if role == "assistant":                # ✅ 标准映射（见下方说明）
        return AIMessage(content=content)
    return HumanMessage(content=content)

# 组装成 W4 那条链：prompt → llm（它就是 Runnable，图节点直接 invoke）
chain = prompt | llm

# ══════════ 2. 定义图：一个节点，壳是图、心是 W4 的链 ══════════
class State(TypedDict):
    question: str    # invoke 时传入
    answer: str      # 由节点填

def run_w4_chain(state: State) -> dict:
    """节点 = 薄壳：取记忆 → 跑 W4 链 → 写回记忆。业务逻辑全在壳外，一行没改"""
    history = [_to_lc(m) for m in mm.get_context(query=state["question"])]
    ai = chain.invoke({"history": history, "question": state["question"]})
    mm.add({"role": "user", "content": state["question"]})          # 记忆照常写回
    mm.add({"role": "assistant", "content": ai.content})
    return {"answer": ai.content}      # 只把 answer 更新进 State

builder = StateGraph(State)
builder.add_node("w4_chain", run_w4_chain)
builder.add_edge(START, "w4_chain")
builder.add_edge("w4_chain", END)
graph = builder.compile()

# ══════════ 3. 跑 ══════════
if __name__ == "__main__":
    print("=== W5-D1 步骤 C：W4 链包成 LangGraph 节点 ===\n")
    print(graph.get_graph().draw_ascii())          # 应看到 START → w4_chain → END

    # 3 连问：①纯问答 ②写记忆 ③考记忆（证明图外组件 mm 照常工作）
    for q in ["你好，简单介绍下你自己",
              "记住：我最喜欢的模型是 DeepSeek",
              "我最喜欢的模型是什么？"]:
        out = graph.invoke({"question": q})        # 每次从 START 全量跑一遍
        print(f"Q: {q}\nA: {out['answer']}\n")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

START → w4_chain → END          ← draw_ascii 画出的图结构
                                  （实际输出是 ASCII 方框排版，这里简化示意）

Q: 你好，简单介绍下你自己
A: 你好！我是有记忆、能调工具的助手……

Q: 记住：我最喜欢的模型是 DeepSeek
A: 好的，我记住了！

Q: 我最喜欢的模型是什么？
A: 你最喜欢的模型是 DeepSeek。   ← 证明 mm 记忆跨 invoke 生效

1
2
3

  第一次 invoke ──▶ mm.add(...) ──▶ 记忆写进 MemoryManager（图外部对象）
  第二次 invoke ──▶ mm.get_context() 读到第一次的  ✅
  第三次 invoke ──▶ mm.get_context() 读到前两次的  ✅

 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

# ===== 四件套 =====
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    question: str                              # 入口字段（invoke 传入）
    answer: str                                # 节点填
    # turns: Annotated[int, operator.add]      # 要"累加"就用 reducer

def node(state: State) -> dict:                # 🎯 必须 return dict（partial update）
    return {"answer": ...}                     # 只返回要更新的字段

builder = StateGraph(State)
builder.add_node("name", node)                 # 加工序
builder.add_edge(START, "name")                # 投料口 → 工序
builder.add_edge("name", END)                  # 工序 → 出口
graph = builder.compile()                      # 🎯 必须 compile 才能 invoke

# ===== 三种跑法 =====
out = graph.invoke({"question": "..."})        # 跑到底，拿最终 State
for chunk in graph.stream({"question": "..."}):# 逐节点观察
    print(chunk)                               # 结构：{节点名: 该节点 return 的 dict}
print(graph.get_graph().draw_ascii())          # 🎯 新版 API，别漏 get_graph()

# ===== 互通：链进图 =====
chain = prompt | llm                           # W4 的 LCEL 链
def run_w4_chain(state: State) -> dict:
    ai = chain.invoke({"history": ..., "question": state["question"]})
    return {"answer": ai.content}              # 只把结果写回 State
builder.add_node("w4_chain", run_w4_chain)

# ===== 三个铁律 =====
# 1. 节点必须 return {...}，不碰全局变量
# 2. 必须 compile() 才能 invoke()
# 3. START / END 从 langgraph.graph 导入

1
2
3
4
5
6
7
8

  今天（D1）：  START → w4_chain → END              线性，无环
  D2：         START → think →┬→ respond → END      有分支
                             └→ retry ↗
  D4：         START → agent →┬→ tools → agent ↗    有环（回边）
                             └→ END

                                    ↑
                          这一刻，while 才真正消失