W4-D6 · 收官：装车

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

┌──────────────────────────────────────────────────────────────────┐
│                     REPL 驾驶舱（while True）                      │
│                     input("用户：") → ask(q) → print              │
└───────────────────────────────┬──────────────────────────────────┘
                                ▼
┌──────────────────────────────────────────────────────────────────┐
│  ask(question) —— 四步闭环（🎯 今天的核心，面试就画这张图）         │
│                                                                    │
│   ① 取记忆   mm.get_context(query=question)                        │
│              └─ 短期窗口 + 长期召回 → dict 列表 → _to_lc 转消息     │
│                             │                                      │
│   ② 拼 prompt  prompt.format_messages(history=..., question=...)   │
│                             │                                      │
│   ③ 生成 + 工具循环  model.invoke() → _run_tools()                 │
│                             │    while ai.tool_calls:              │
│                             │      执行 → ToolMessage 回填 → 再问   │
│                             │                                      │
│   ④ 写回记忆  mm.add(user) + mm.add(assistant)                     │
└──────────────────────────────────────────────────────────────────┘

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

用户输入
  ↓
① 取记忆：mm.get_context(query)          ← D3：短期窗口 + 长期召回，dict → BaseMessage
  ↓
② 拼 prompt：历史经 placeholder 展开 + 当前问题   ← D1 / D3
  ↓
③ 模型生成 + 工具循环：
     while ai.tool_calls → 执行 → ToolMessage 回填 → 再问   ← D5
  ↓
④ 写回记忆：mm.add(user) + mm.add(assistant)   ← D3：只有 ask() 能写回
  ↓
输出回答 → 回到 ①（下一轮）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

  ┌─────────────────────────────────────────────────┐
  │  链（无状态）                                     │
  │  prompt | model | parser                         │
  │  每次 invoke 都是白纸一张，不带任何上轮痕迹        │
  └─────────────────────────────────────────────────┘
              ↕ 需要一道"人工搬运"
  ┌─────────────────────────────────────────────────┐
  │  记忆（有状态）                                   │
  │  MemoryManager：短期窗口 + 长期 Chroma + 压缩     │
  │  跨轮次持续存在                                   │
  └─────────────────────────────────────────────────┘

1
2
3
4
5

import os, sys
from pathlib import Path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))  # 回到根目录，硬编码
from dotenv import load_dotenv
load_dotenv(Path(__file__).parent.parent / ".env", override=True)

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

model = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    temperature=0,
).bind_tools(tools, parallel_tool_calls=False)   # ← 注意这个参数

mm = MemoryManager(window=8, max_tokens=1500, keep=4)

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

1
2
3
4
5
6
7

def _to_lc(m: dict):
    role, content = m.get("role", "user"), m.get("content", "")
    if role == "system":
        return SystemMessage(content=content)
    if role == "user":
        return AIMessage(content=content)      # ⚠️ 这一行有问题，见坑 5
    return HumanMessage(content=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

MAX_ROUNDES = 3

def _normalize_args(args: dict) -> dict:
    """任意层嵌套都取第一个字符串值"""
    for k, v in list(args.items()):
        while isinstance(v, dict) and v:
            v = next(iter((v.values)))         # ⚠️ 漏了括号，见坑 6
        args[k] = v
    return args

def _run_tools(messages, verbose=True):
    """串行执行：一次只处理一个 tool_call，最稳的 DeepSeek 兼容方案"""
    round_ = 0
    ai = messages[-1]
    while ai.tool_calls and round_ < MAX_ROUNDES:
        round_ += 1
        if verbose:
            print(f"  [工具第{round_}轮] 模型发起 {len(ai.tool_calls)} 个调用:")
            for call in ai.tool_calls:
                print(f"    → {call['name']}({_normalize_args(call['args'])})")
        for call in ai.tool_calls:                        # 注意：不是 tool_calls[0]！
            args = _normalize_args(call.get("args", {}))
            try:
                result = tools_by_name[call['name']].invoke(args)
                content = str(result) if result is not None else "无结果"
            except Exception as e:
                content = f"工具调用失败: {e}"             # 失败也必须回填，否则协议违例
            messages.append(ToolMessage(content=content, tool_call_id=call['id']))
        ai = model.invoke(messages)                   # 模型看到结果后，自己决定还要不要发下一个
        messages.append(ai)
    return ai

1
2
3
4
5
6

try:
    result = tools_by_name[call['name']].invoke(args)
    content = str(result) if result is not None else "无结果"
except Exception as e:
    content = f"工具调用失败: {e}"
messages.append(ToolMessage(content=content, tool_call_id=call['id']))

1
2
3
4
5
6
7
8

  模型: 我要调 get_weather（tool_call_id = abc123）
  你:   调失败了，跳过不回填 ❌
  下一轮 invoke(messages) → 400 Bad Request
        ↑ messages 里有个没有结果的 tool_call，协议不完整

  正确做法：
  你:   调失败，但照样回填 ToolMessage(content="工具调用失败: ...", id=abc123) ✅
  下一轮 invoke → 模型看到失败信息，自己决定换个说法或改调别的工具

1
2
3
4
5
6
7
8
9

def ask(question: str) -> str:
    history = [_to_lc(m) for m in mm.get_context(query=question)]        # ① 取记忆
    messages = prompt.format_messages(history=history, question=question)  # ② 拼 prompt
    ai = model.invoke(messages)
    messages.append(ai)
    ai = _run_tools(messages)                                            # ③ 生成 + 工具循环
    mm.add({"role": "user", "content": question})                        # ④ 写回记忆
    mm.add({"role": "assistant", "content": ai.content})
    return ai.content

1
2
3
4

  记忆里存的：                        记忆里不存的：
  ├─ user: "北京天气怎么样？"          ├─ AIMessage(tool_calls=[get_weather])
  └─ assistant: "北京今天晴，26℃"      └─ ToolMessage(content="北京今天晴，26℃")
                                        ↑ 这些是过程，存在 messages 里就够了

1
2
3
4
5
6
7

if __name__ == "__main__":
    print("=== W4-D6：带记忆 + 工具的多轮对话 demo ===")
    while True:
        question = input("\n用户：")
        if question in ["exit", "quit"]:
            break
        print("助手:", ask(question))

1
2
3
4
5
6
7
8

  用户："我刚才说了什么？"
     │
     ├─ 短期通道：history 里已经有上一轮内容 ──▶ 直接答 ✅（无需调工具）
     │
  用户："用户喜欢什么模型，预算多少？"
     │
     ├─ 短期通道：history 里没有（这是几轮前/上次会话的事）
     ├─ 模型判断：得翻档案 ──▶ 调 memory_retriever ──▶ 长期通道命中 ✅

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

# W4 收官 Demo：带记忆 + 工具的对话 Agent

## 一句话
D1~D5 的零件装成的多轮对话闭环：记忆（短期+长期）→ 模型 → 三工具 → 回写记忆。

## 依赖
pip install langchain langchain-openai langchain-chroma python-dotenv tavily-python

## 环境变量（放项目根目录 .env，勿硬编码）
DEEPSEEK_API_KEY=sk-xxx
TAVILY_API_KEY=tvly-xxx

## 运行
python W4/lc_agent_demo.py

## 零件清单
- lc_tools_multi.py  → D5 三工具（天气/搜索/长期记忆）
- retriever.py       → D4 MemoryRetriever（官方认证检索器）
- memory/            → W3 记忆门面（短期窗口+长期Chroma+压缩）

1
2
3
4
5

  测试 1 挂 → 模型/密钥层（连不上）
  测试 2 挂 → _to_lc 或 get_context（记忆没取到 / 角色错了）← 坑 5 会挂在这里
  测试 3 挂 → bind_tools 或工具循环（D5 没通）
  测试 4 挂 → retriever 或 memory_tool（D4 没通 / docstring 没写清）
  测试 5 挂 → REPL 层（最不可能，但得验）

1
2
3
4
5
6
7

def _to_lc(m: dict):
    role, content = m.get("role", "user"), m.get("content", "")
    if role == "system":
        return SystemMessage(content=content)
    if role == "user":
        return AIMessage(content=content)      # ❌ user 被映射成 AIMessage
    return HumanMessage(content=content)       # ❌ assistant 落到这里变成 HumanMessage

1
2
3
4
5
6
7
8

  实际发生的事：
  你说的 "我叫小明"      → AIMessage   → 模型以为是"自己说过的话"
  模型说的 "你好小明"    → HumanMessage → 模型以为是"用户说的话"

  多轮对话会怎么乱：
  第1轮 你：我叫小明        → [AI] 我叫小明
  第2轮 你：我叫什么？      → 模型看历史，发现"小明"是自己说的
                            → 它可能回答"我没说过你叫什么"

1
2
3
4
5
6
7

def _to_lc(m: dict):
    role, content = m.get("role", "user"), m.get("content", "")
    if role == "system":
        return SystemMessage(content=content)
    if role == "assistant":                    # ✅ 改成 assistant
        return AIMessage(content=content)
    return HumanMessage(content=content)       # ✅ user 落到这里

1
2

while isinstance(v, dict) and v:
    v = next(iter((v.values)))      # ❌ 少了调用括号

1
2

while isinstance(v, dict) and v:
    v = next(iter(v.values()))      # ✅ 加括号调用

1

TypeError: 'builtin_function_or_method' object is not iterable

 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

# 1. 模型绑定工具（D5）
model = ChatOpenAI(model=..., api_key=..., base_url=..., temperature=0) \
            .bind_tools(tools, parallel_tool_calls=False)   # D6 加固：DeepSeek 串行更稳

# 2. 工具声明（D5）
@tool
def get_weather(city: str) -> str:
    """查询天气。参数 city: 城市名。"""     # docstring 就是给模型看的说明书
    return f"{city} 今天晴, 26℃"

# 3. 工具查表 + 循环回填（D5 核心，D6 加固）
tools_by_name = {t.name: t for t in tools}
while ai.tool_calls and round_ < MAX_ROUNDS:
    for call in ai.tool_calls:
        try:
            result = tools_by_name[call["name"]].invoke(_normalize_args(call.get("args", {})))
            content = str(result) if result is not None else "无结果"
        except Exception as e:
            content = f"工具调用失败: {e}"      # 🎯 失败也必须回填
        messages.append(ToolMessage(content=content, tool_call_id=call["id"]))
    ai = model.invoke(messages)
    messages.append(ai)

# 4. prompt 带历史占位（D1/D3）
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是有记忆、能调工具的助手。"),
    ("placeholder", "{history}"),          # 吃消息对象列表
    ("human", "{question}"),
])

# 5. 消息桥（D3）—— 注意 assistant 的映射别写反
def _to_lc(m: dict):
    if m.get("role") == "system":    return SystemMessage(content=m["content"])
    if m.get("role") == "assistant": return AIMessage(content=m["content"])
    return HumanMessage(content=m["content"])

# 6. 参数归一化（D5 踩坑修复）—— 注意 v.values() 的括号
def _normalize_args(args: dict) -> dict:
    for k, v in list(args.items()):
        while isinstance(v, dict) and v:
            v = next(iter(v.values()))
        args[k] = v
    return args

# 7. 闭环四步（D6 灵魂）
def ask(question: str) -> str:
    history = [_to_lc(m) for m in mm.get_context(query=question)]      # ① 取（query 不能省）
    messages = prompt.format_messages(history=history, question=question)  # ② 拼
    ai = model.invoke(messages); messages.append(ai)
    ai = _run_tools(messages)                                          # ③ 工具循环
    mm.add({"role": "user", "content": question})                      # ④ 写回
    mm.add({"role": "assistant", "content": ai.content})
    return ai.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
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

# ============ W4 收官自测：从零默写"带记忆 + 工具的 Agent" ============
import os, sys
from pathlib import Path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from dotenv import load_dotenv
load_dotenv(Path(__file__).parent.parent / ".env", override=True)

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_core.retrievers import BaseRetriever
from langchain.tools.retriever import create_retriever_tool

# ① 模型（D5）—— 要写：bind_tools + parallel_tool_calls
model = ChatOpenAI(                      # ← 填空：model / api_key / base_url / temperature
    ...
).bind_tools(tools, parallel_tool_calls=False)

# ② 工具1：天气（D5）—— 要写：@tool + docstring
@tool
def get_weather(city: str) -> str:
    """..."""                            # ← 填空：给模型的说明书
    return f"{city} 今天晴, 26℃"

# ③ 工具2：搜索（D5 闭包工厂）—— 要写：内部 @tool
def make_web_search_tool(api_key: str):
    @tool
    def search_web(query: str, max_results: int = 3) -> str:
        """..."""
        pass                             # ← 填空：调用搜索 API
    return search_web

# ④ 工具3：记忆检索（D4）—— 要写：create_retriever_tool
def make_memory_tool(retriever: BaseRetriever):
    return create_retriever_tool(...)    # ← 填空：retriever / name / description

# ⑤ 查表（D5）
tools = [...]                            # ← 填空：三个工具
tools_by_name = {t.name: t for t in tools}

# ⑥ prompt（D1/D3）—— 要写：placeholder
prompt = ChatPromptTemplate.from_messages([...])

# ⑦ 消息桥（D3）—— 注意别把 assistant 写成 user
def _to_lc(m: dict):
    pass

# ⑧ 参数归一化（D5 踩坑）—— 注意 v.values() 括号
def _normalize_args(args: dict) -> dict:
    pass

# ⑨ 工具循环（D5 核心）—— while + 上限 + 失败也回填
MAX_ROUNDS = 3
def _run_tools(messages, verbose=True):
    pass

# ⑩ 核心 ask()（D6 收官）—— 四步闭环
def ask(question: str) -> str:
    pass

# ⑪ REPL 驾驶舱（D6）
if __name__ == "__main__":
    while True:
        q = input("\n用户：").strip()
        if q in ("exit", "quit"):
            break
        print("助手:", ask(q))