W4-D5 · Retriever 转正 + Tool 调用

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

┌──────────────────────────────────────────────────────────────┐
│  第一块  D4 收尾：临时工 → 正式工                              │
│                                                                │
│  RunnableLambda 手工包          ──▶   BaseRetriever 子类        │
│  （能用，但没工牌）                    （活一样干，门禁全开）    │
├──────────────────────────────────────────────────────────────┤
│  第二块  D5 正篇：Tool & Tool Calling                          │
│                                                                │
│  W2 手写的 fc_loop              ──▶   @tool + bind_tools        │
│  （自己写 schema、自己写循环）         （schema 自动生成，        │
│                                         循环逻辑原样保留）      │
└──────────────────────────────────────────────────────────────┘

1
2
3
4
5
6
7
8
9

def make_retriever(mm, k=3):
    """把 mm.recall 包成 LangChain 标准 Retriever：query -> list[Document]"""
    def recall_docs(query: str) -> list[Document]:
        facts = mm.recall(query, k=k)
        return [Document(page_content=f, metadata={"source": "lang_term"}) for f in facts]
    return RunnableLambda(recall_docs)

retriever = make_retriever(mm, k=3)
docs = retriever.invoke("用户喜欢什么模型？")

1
2
3
4
5
6

chain = (
    {"context": retriever | _ftm, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

1
2
3
4
5
6
7
8

临时工 RunnableLambda              正式工 BaseRetriever
─────────────────────              ─────────────────────
框架眼里：一个普通函数               框架眼里：一个检索器
  · 能 invoke            ✅           · 能 invoke               ✅
  · 能进链               ✅           · 能进链                  ✅
  · create_retriever_tool ❌ 不认      · create_retriever_tool    ✅ 认
  · LangGraph 检索节点    ❌ 不认      · LangGraph 检索节点       ✅ 认
  · RetrievalQA          ❌ 不认      · RetrievalQA             ✅ 认

 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

# BaseRetriver 子类
from typing import Any
from langchain_core.retrievers import BaseRetriever
from langchain_core.documents import Document
from pydantic import Field, ConfigDict


class MemoryRetriever(BaseRetriever):
    """把 W3 的 Memory Manager.recall 包装成官方的 Retriever
       BaseRetriever 是个 pydantic 模型，所以字段要声明类型 + Field
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)  # 允许非 pydantic 对象
    mm: Any = Field(description="带 recall(query, k) 方法的记忆管理器")
    k: int = Field(default=3, description="召回条数")

    def _get_relevant_documents(self, query: str, **kwargs) -> list[Document]:
        """唯一必须实现的抽象方法：query → Document 列表。

        **kwargs 兼容新旧版本签名（旧版无 run_manager，新版有）。
        """
        facts = self.mm.recall(query, k=self.k)          # W3 的召回
        return [
            Document(page_content=f, metadata={"source": "长期记忆"})
            for f in facts
        ]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

  retriever.invoke("用户喜欢什么颜色")
            │
            ▼
  BaseRetriever.invoke()          ← 框架自带：参数校验、回调、配置注入
            │
            ▼
  _get_relevant_documents(query)  ← 🎯 只有这一块是你写的
            │
            ▼
  [Document, Document, Document]

1
2

docs = retriever.invoke("用户喜欢什么颜色")       # list[Document]
docs_list = retriever.batch(["问题1", "问题2"])   # 批量，框架自动并发

1
2
3
4
5
6

from langchain_core.runnables import RunnableLambda

chain = retriever | RunnableLambda(
    lambda docs: "；".join(d.page_content for d in docs)
)
result = chain.invoke("用户喜欢什么模型？")   # 返回纯字符串

1
2
3
4
5
6
7

普通类                          pydantic 模型（BaseRetriever）
─────────                       ────────────────────────────
self.mm = mm                    mm: Any = Field(description="...")
self.k = 3                      k: int = Field(default=3, ...)
（__init__ 里随便赋值）          （类级别声明类型 + 元数据）
不做校验                         实例化时校验类型
不能序列化                       能 .model_dump() 存下来

1
2

pydantic.errors.PydanticSchemaGenerationError:
  Unable to generate pydantic-core schema for <class 'memory.memory_manager.MemoryManager'>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 之前（临时工）:
# make_retriever = RunnableLambda(lambda q: [Document(page_content=f) for f in mm.recall(q)])

# 现在（正式工）:
from W4.retriever import MemoryRetriever

mm = MemoryManager(window=8, max_tokens=1500, keep=4)   # 不需要 llm，纯召回
retriever = MemoryRetriever(mm=mm, k=3)

# 用法完全一样，甚至更简单：
docs = retriever.invoke("用户喜欢什么颜色")

1
2
3
4
5
6
7

# 原来：retriever = make_retriever(mm, k=3)
# 现在：retriever = MemoryRetriever(mm=mm, k=3)
# 下面这段原封不动
chain = (
    {"context": retriever | _ftm, "question": RunnablePassthrough()}
    | prompt | model | StrOutputParser()
)

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

# test_retriever.py
from W4.retriever import MemoryRetriever

class FakeMemory:
    """模拟 MemoryManager.recall —— 不联网也能测"""
    def recall(self, query, k=3):
        return ["用户叫小明", "喜欢蓝色", "预算2000"]

r = MemoryRetriever(mm=FakeMemory(), k=3)

# 测试1：invoke 返回 list[Document]
docs = r.invoke("用户喜欢什么")
assert isinstance(docs, list) and len(docs) == 3
print(f"✓ invoke 返回 {len(docs)} 条 Document")

# 测试2：page_content 和 metadata 都在
print(f"✓ 内容: {docs[1].page_content} | 来源: {docs[1].metadata['source']}")

# 测试3：能直接进 LCEL 链
from langchain_core.runnables import RunnableLambda
chain = r | RunnableLambda(lambda ds: "；".join(d.page_content for d in ds))
print(f"✓ 链输出: {chain.invoke('用户喜欢什么')}")

1
2
3

✓ invoke 返回 3 条 Document
✓ 内容: 喜欢蓝色 | 来源: 长期记忆
✓ 链输出: 用户叫小明；喜欢蓝色；预算2000

1
2
3
4

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气。参数 city: 城市名。"""
    return f"{city} 今天晴，26℃"

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

  你 W2 的手工活                      D5 的自动活
  ──────────────                      ────────────
  TOOLS_SCHEMA = [{                   def get_weather(city: str):
      "name": "get_weather",              """查询指定城市的天气。
      "description": "查询天气",             参数 city: 城市名。"""
      "parameters": {                           │
          "type": "object",                     │  @tool 自动生成
          "properties": {                       ▼
              "city": {"type": "string",    name="get_weather"
                       "description": ...}  description="查询指定城市的天气..."
          },                                args_schema={city: str}
          "required": ["city"]
      }
  }]

1

model = ChatOpenAI(...).bind_tools(tools)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

  model = ChatOpenAI(...)                    ← 普通模型：只能说话
           │
           │  .bind_tools(tools)
           ▼
  model_with_tools                           ← 带登记簿的模型：能说话，也能填单
           │
           │  .invoke(messages)
           ▼
  AIMessage(content="", tool_calls=[...])    ← 想调工具时，content 为空、tool_calls 有货
  或
  AIMessage(content="北京今天晴", tool_calls=[])  ← 不用工具时，直接给答案

1
2
3
4
5
6
7

ai = model.invoke(messages)          # ① 让模型回答
while ai.tool_calls:                 # ② 模型要调工具？
    for call in ai.tool_calls:
        result = tools_by_name[call["name"]].invoke(call["args"])              # ③ 执行工具
        messages.append(ToolMessage(content=str(result), tool_call_id=call["id"]))  # ④ 回填
    ai = model.invoke(messages)      # ⑤ 带着结果再问模型
return ai.content                    # ⑥ 无 tool_calls → 最终答案

 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

┌──────────────────────────────────────────────────────────────────┐
│                     W4/lc_tools_multi.py                          │
│                                                                    │
│   【三个工具】                                                      │
│   ┌─────────────────┐ ┌─────────────────┐ ┌──────────────────┐  │
│   │ get_weather     │ │ search_web      │ │ memory_tool      │  │
│   │ @tool 直出      │ │ 工厂函数 + 闭包  │ │ 由 D4 正式工      │  │
│   │ 无状态，最简单   │ │ 注入 TavilyClient│ │ retriever 转换来  │  │
│   └────────┬────────┘ └────────┬────────┘ └────────┬─────────┘  │
│            └───────────────────┼───────────────────┘              │
│                                ▼                                   │
│                     tools = [三个工具]                             │
│                     tools_by_name = {名字: 工具}                   │
│                                │                                   │
│                                ▼                                   │
│                  model.bind_tools(tools)                           │
│                                │                                   │
│                                ▼                                   │
│   ┌────────────────────────────────────────────────────────────┐ │
│   │  run(query) —— 手写循环                                      │ │
│   │  while ai.tool_calls and round_ < MAX_ROUNDS:               │ │
│   │      args = _normalize_args(call["args"])   ← 参数归一化     │ │
│   │      result = tools_by_name[名字].invoke(args)              │ │
│   │      messages.append(ToolMessage(...))                       │ │
│   └────────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────────┘

1
2
3
4

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气。参数 city: 城市名。"""
    return f"{city} 今天晴，26℃"

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

def make_web_search_tool(client=None):
    """工厂函数：把有状态的 client 闭包捕获，返回 @tool 工具"""
    if client is None:
        from tavily import TavilyClient
        client = TavilyClient(api_key=os.environ.get("TAVILY_API_KEY"))

    @tool
    def search_web(query: str, max_results: int = 3) -> str:
        """搜索网页信息。当用户询问需要联网查询的问题、最新动态时，调用此工具。
           参数 query: 搜索关键词; max_results: 返回条数(默认三条)"""
        result = client.search(query, max_results=max_results)
        return json.dumps(result, ensure_ascii=False)

    return search_web

1
2
3
4
5
6
7
8
9

  第一层：make_web_search_tool(client)      ← 工厂：负责"生"工具
              │
              │  client 被闭包捕获（不在 @tool 的签名里）
              ▼
  第二层：@tool def search_web(query, max_results)  ← 模型只看见这两个参数
              │
              │  函数体内部直接用外层的 client
              ▼
  第三层：return search_web                  ← 交出一个"带着 client 的工具"

1
2
3
4
5
6
7
8

  ❌ 错误想法：把 client 当参数
     @tool
     def search_web(client, query): ...     # 模型不知道 client 该填什么

  ✅ 正确做法：闭包在"定义时"注入，模型只在"调用时"填业务参数
     def make_web_search_tool(client):      # 定义时注入依赖
         @tool
         def search_web(query): ...         # 调用时只需业务参数

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

from W4.retriever import MemoryRetriever
try:
    from langchain.tools.retriever import create_retriever_tool
except ImportError:
    from langchain_core.tools.retriever import create_retriever_tool

mm = MemoryManager(window=8, max_tokens=1500, keep=4)
mm.add({"role": "user", "content": "用户喜欢Deekseek，预算2000元"}, persist=True)
mm.add({"role": "user", "content": "用户叫小明，喜欢蓝色"}, persist=True)

memory_retriever = MemoryRetriever(mm=mm, k=3)
memory_tool = create_retriever_tool(
    memory_retriever,
    name="memory_retriever",
    description="从长期记忆检索用户偏好、历史事实。当用户问起以前说过/聊过的事情时调用"
)

1
2
3

search_web = make_web_search_tool()          # 闭包捕获 TavilyClient
tools = [get_weather, search_web, memory_tool]
tools_by_name = {t.name: t for t in tools}

1
2
3
4
5
6

model = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    temperature=0,                    # ← 工具调用场景建议 0：要稳定，不要发散
).bind_tools(tools)

 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

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

MAX_ROUNDS = 3

def run(query: str, verbose: bool = True):
    messages = [HumanMessage(query)]
    ai = model.invoke(messages)
    messages.append(ai)

    round_ = 0
    while ai.tool_calls and round_ < MAX_ROUNDS:
        round_ += 1

        if verbose:
            print(f"\n[第{round_}轮] 模型发起 {len(ai.tool_calls)} 个工具调用:")
            for call in ai.tool_calls:
                print(f"  工具: {call['name']}  参数: {call['args']}  解析后: {_normalize_args(call['args'])}")

        for call in ai.tool_calls:
            args = _normalize_args(call["args"])
            result = tools_by_name[call["name"]].invoke(args)
            messages.append(ToolMessage(content=str(result), tool_call_id=call["id"]))
            if verbose:
                print(f"  工具返回 {str(result)[:80]}...")

        ai = model.invoke(messages)          # 带着结果再问模型
        messages.append(ai)

    if ai.tool_calls:
        print(f"\n[第{round_}轮] 模型发起 {len(ai.tool_calls)} 个工具调用，"
              f"但已达到最大轮数 {MAX_ROUNDS}，停止调用。")
    return ai.content

1
2
3
4
5

  裸 while ai.tool_calls:            带 MAX_ROUNDS 的循环
  ─────────────────────             ────────────────────
  ✅ 逻辑正确                         ✅ 逻辑正确
  ❌ 可能无限循环烧钱                  ✅ 最多 3 轮，强制退出
  ❌ 生产环境不敢上线                  ✅ 有兜底返回值

1

{"query": {"description": "搜索关键词", "type": "string"}}

1
2
3
4
5
6

def _normalize_args(args: dict) -> dict:
    for k, v in list(args.items()):
        while isinstance(v, dict) and v:      # 只要还是非空 dict，就继续剥
            v = next(iter(v.values()))        # 取第一个 value 接着剥
        args[k] = v
    return args

1
2
3
4
5
6

  输入: {"query": {"description": "搜索关键词", "type": "string"}}
         │
         │  isinstance(v, dict) and v  → True
         │  v = next(iter(v.values())) = "搜索关键词"
         ▼
  输出: {"query": "搜索关键词"}          ← 剥到非 dict 就停

1

print(f"  工具: {call['name']}  参数: {call['args']}  解析后: {_normalize_args(call['args'])}")

1
2
3
4
5
6
7

if __name__ == "__main__":
    print("\n=== 测试1：天气 ===")
    print(run("北京天气怎么样?"))
    print("\n=== 测试2：联网搜索 ===")
    print(run("今天的最新AI新闻是什么?"))
    print("\n========== 测试3：长期记忆 ==========")
    print(run("用户喜欢什么模型，预算多少?"))

1
2
3
4

try:
    from langchain.tools.retriever import create_retriever_tool
except ImportError:
    from langchain_core.tools.retriever import create_retriever_tool

1
2
3
4

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
...
project_root = Path(__file__).parent.parent
load_dotenv(project_root / ".env", override=True)

1
2
3

model = ChatOpenAI(...)           # 这个 model 没有工具
model.bind_tools(tools)           # ❌ 返回值被丢弃，等于没绑
model = ChatOpenAI(...).bind_tools(tools)   # ✅ 接住返回值

 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

# ===== D4：Retriever 转正 =====
from langchain_core.retrievers import BaseRetriever
from langchain_core.documents import Document
from pydantic import Field, ConfigDict

class MemoryRetriever(BaseRetriever):
    model_config = ConfigDict(arbitrary_types_allowed=True)   # 放行非 pydantic 对象
    mm: Any = Field(description="带 recall(query, k) 的记忆管理器")
    k: int = Field(default=3, description="召回条数")

    def _get_relevant_documents(self, query: str, **kwargs) -> list[Document]:
        # 🎯 唯一必须实现的方法；**kwargs 兼容版本差异
        return [Document(page_content=f, metadata={"source": "长期记忆"})
                for f in self.mm.recall(query, k=self.k)]

# ===== D5：工具三件套 =====
from langchain_core.tools import tool

@tool
def my_tool(city: str) -> str:
    """干什么。什么时候调用。参数 city: 城市名。"""   # ← docstring = 说明书
    return "..."

def make_stateful_tool(client):          # 有状态依赖：工厂 + 闭包
    @tool
    def inner(query: str) -> str:
        """..."""
        return client.search(query)      # client 从闭包来
    return inner

model = ChatOpenAI(...).bind_tools(tools)          # ⚠️ 记得接住返回值
tools_by_name = {t.name: t for t in tools}         # 名字 → 工具对象的查表

# ===== 循环 =====
ai = model.invoke(messages)
while ai.tool_calls and round_ < MAX_ROUNDS:       # 🎯 必须有界
    for call in ai.tool_calls:
        args = _normalize_args(call["args"])                        # 剥嵌套 dict
        result = tools_by_name[call["name"]].invoke(args)
        messages.append(ToolMessage(content=str(result), tool_call_id=call["id"]))
    ai = model.invoke(messages)                    # 带结果再问
return ai.content

# ===== retriever → tool（转正的红利）=====
try:
    from langchain.tools.retriever import create_retriever_tool
except ImportError:
    from langchain_core.tools.retriever import create_retriever_tool
memory_tool = create_retriever_tool(memory_retriever, name="memory_retriever",
                                    description="从长期记忆检索……当用户问起以前聊过的事时调用")