1
2
3
4
5
6
7

用户输入 {"question": "我叫什么？"}
   ↓
load_mem (RunnableLambda)   ← ① 给输入"加料"：补上 history 字段
   ↓
prompt (ChatPromptTemplate) ← ② {history} 和 {question} 这下都有值了
   ↓
model → parser              ← ③ 生成回答

1
2

[{"role": "user", "content": "我叫小明"},
 {"role": "assistant", "content": "你好小明"}]

1
2
3
4
5
6
7

def _to_lc(m: dict):
    role, content = m.get("role", "user"), m.get("content", "")   # 取出角色和内容，role 缺省默认 user
    if role == "system":
        return SystemMessage(content=content)       # system → SystemMessage
    if role == "assistant":
        return AIMessage(content=content)           # assistant → AIMessage
    return HumanMessage(content=content)            # 其他（user）→ HumanMessage

1
2
3
4
5
6
7

# 你的代码里：
lambda x: {**x, "history": get_history(x)}
#    ↑参数    ↑ 返回表达式（一个字典）

# 它等价于这个 def：
def 匿名函数(x):
    return {**x, "history": get_history(x)}

1
2
3
4

x = {"question": "我叫什么？"}

{**x, "history": ["..."]}
# 展开 x → {"question": "我叫什么？", "history": ["..."]}

1
2
3
4
5
6

# ❌ 直接放普通函数：
chain = get_history | prompt | model | parser   # 报错：get_history 不是 Runnable

# ✅ 包一层 RunnableLambda：
load_mem = RunnableLambda(get_history)
chain = load_mem | prompt | model | parser      # 能跑！

 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

# concepts_demo.py —— 三个概念一次看懂（无需 API key）
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langchain_core.runnables import RunnableLambda

# ---------- 1. _to_lc：dict → 消息对象 ----------
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)

raw = [{"role": "user", "content": "我叫小明"},
       {"role": "assistant", "content": "你好小明"}]
msgs = [_to_lc(r) for r in raw]
for m in msgs:
    print(type(m).__name__, "→", m.content)
# HumanMessage → 我叫小明
# AIMessage → 你好小明

# ---------- 2. lambda 语法 ----------
f = lambda x: x + "！"               # lambda 参数: 返回表达式
def f_def(x): return x + "！"        # 等价 def
print(f("你好"), "==", f_def("你好"))  # 你好！ == 你好！

# ---------- 3. RunnableLambda + 字典解包 ----------
def get_history(x):
    return [_to_lc(r) for r in raw]   # 返回消息对象列表（就是 _to_lc 的用武之地）

load_mem = RunnableLambda(lambda x: {**x, "history": get_history(x)})

result = load_mem.invoke({"question": "我叫什么？"})
print(result.keys())     # dict_keys(['question', 'history'])
print(result["history"]) # [HumanMessage, AIMessage] —— prompt 的 {history} 就能展开了

 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
75
76
77
78
79
80
81
82
83

# lc_memory.py —— W3 MemoryManager 接进 LCEL 链（入口适配层）
import os
from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

load_dotenv()

# 桥的包装器：dict 消息 → LangChain 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)

# 接入 W3 自研 MemoryManager（失败则降级最小窗口版）
# ⚠️ 注意：直接 `python lc_memory.py` 运行时，下面这行相对导入会找不到包——
#    因为直接运行脚本时 Python 把它当顶层模块，无法解析 "..memory" 这类上级路径。
#    正确做法（二选一）：
#      1) 把项目根目录加入 sys.path 再用绝对导入（见下方注释）；
#      2) 用 `python -m 包名.模块名` 启动，而非直接执行脚本文件。
try:
    from memory.memory_manager import MemoryManager
    _mm = MemoryManager(window=6, max_tokens=500)
    def get_history(x):
        # 真实版：query=本轮问题 → 长期召回 + 滚动窗口
        return [_to_lc(m) for m in _mm.get_context(query=x.get("question"))]
except Exception as e:
    print(f"[降级]未找到memory包({e}),使用最小窗口版")
    class _MinilMM:
        def __init__(self, window=6):
            self.buf, self.window = [], window
        def get_context(self, query=None):
            return self.buf[-self.window:]
        def add_user(self, t):
            return self.buf.append({"role": "user", "content": t})
        def add_ai(self, t):
            self.buf.append({"role": "assistant", "content": t})
    _mm = _MinilMM()
    def get_history(x):
        return [_to_lc(m) for m in _mm.get_context()]

# 链：链首选历史，再 prompt | model | parser
model = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有记忆的助手，记得之前聊过的内容。"),
    ("placeholder", "{history}"),            # 关键：展开历史 messages
    ("human", "{question}"),
])

load_mem = RunnableLambda(lambda x: {**x, "history": get_history(x)})
chain = load_mem | prompt | model | StrOutputParser()

def ask(question: str) -> str:
    ans = chain.invoke({"question": question})
    if hasattr(_mm, "add_user"):
        _mm.add_user(question); _mm.add_ai(ans)
    else:
        _mm.add({"role": "user", "content": question})
        _mm.add({"role": "assistant", "content": ans})
    return ans

if __name__ == "__main__":
    # 验收：第 2、3 问能答出「篮球」「小明」=> 记忆生效
    for q in ["我叫小明，喜欢打篮球", "我刚说的爱好是什么？", "我名字叫什么？"]:
        print(f"Q: {q}\nA: {ask(q)}\n")

    # 确认 load_mem 在链首 + 看历史在累积
    print("=== 链结构图（load_mem 应排最前）===")
    chain.get_graph().print_ascii()
    print("=== 当前记忆历史 ===")
    for m in _mm.get_context():
        print(f"  {m.get('role')}: {m.get('content')}")