Langchain学习笔记1-管道符|构建链路问题初探
问题
学习摘要记忆时，下面一段代码不太理解：变量x就是上一轮的输出吗？那第一次是怎么执行的？
|
首先搞清| 的原理，Runnable重写了__or__，
继续点开函数coerce_to_runnable，
# 源码 def coerce_to_runnable(thing: RunnableLike) -> Runnable[Input, Output]: if isinstance(thing, Runnable): return thing if is_async_generator(thing) or inspect.isgeneratorfunction(thing): return RunnableGenerator(thing) if callable(thing): return RunnableLambda(cast("Callable[[Input], Output]", thing)) # 注意这里，dict会转为Runnable实例 if isinstance(thing, dict): return cast("Runnable[Input, Output]", RunnableParallel(thing)) msg = ( f"Expected a Runnable, callable or dict." f"Instead got an unsupported type: {type(thing)}" ) raise TypeError(msg)可以看到字典转化为了Runnable实例，因此字典跟Runnable实例可以直接使用|操作符，测试下面两行代码执行通过。
{}|summary_prompt{}|llm第一次invoke
按照人正常工作的逻辑，第一次invoke用不到history。那summary_base_chain怎么处理历史记忆的，于是改一下代码，将lambda替换为外部定义的函数，打印内部过程。
# 生成摘要的prompt，输入为上一轮对话的输出，输出的摘要文本做为输入加入historysummary_prompt=ChatPromptTemplate.from_messages([("system","你是对话摘要助手，需简洁总结以下对话的核心信息（包含用户身份、偏好、关键问题等），不超过50字。"),("human","对话历史：{chat_history_text}\n请生成摘要：")])summary_chain=summary_prompt|llm summary_memory_prompt=ChatPromptTemplate.from_messages([("system","你是友好的对话助手，需基于对话摘要回答用户问题，摘要包含核心上下文信息。"),("system","对话摘要：{chat_summary}"),# 注入摘要("human","{user_input}")])# 4. 构建基础对话链（提示词 + LLM）deff(x:dict):print("摘要以下输入x=",x)res=summary_chain.invoke({"chat_history_text":"\n".join([f"{msg.type}:{msg.content}"formsginx["chat_history"]]# x["chat_history"]为空列表则invoke空字符串)})c=res.contentprint(f"摘要输出content={c}",f'摘要消耗total_tokens={res.response_metadata['token_usage']['total_tokens']}')returnc summary_base_chain=(RunnablePassthrough.assign(chat_summary=f)|summary_memory_prompt|llm)# 5. 会话历史存储（保存完整历史用于生成摘要）summary_memory_store={}# 6. 定义会话历史获取函数defget_summary_memory_history(session_id:str)->BaseChatMessageHistory:ifsession_idnotinsummary_memory_store:summary_memory_store[session_id]=InMemoryChatMessageHistory()returnsummary_memory_store[session_id]# 7. 构建带摘要记忆的对话链summary_memory_chain=RunnableWithMessageHistory(runnable=summary_base_chain,get_session_history=get_summary_memory_history,input_messages_key="user_input",history_messages_key="chat_history"# 传入完整历史用于生成摘要)# 测试多轮对话（session_id=user_003）config={"configurable":{"session_id":"user_003"}}# 多轮对话输入inputs=["我叫小李，是一名产品经理","我负责一款电商APP的迭代",# "最近在优化用户下单流程",# "遇到了用户流失率高的问题",# "你能给我一些优化建议吗？"]fori,user_inputinenumerate(inputs,1):response=summary_memory_chain.invoke({"user_input":user_input},config=config)print(f"\n第{i}轮 - 助手回复：",response.content)代码片打印结果：
"""摘要以下输入x={'user_input':'我叫小李，是一名产品经理','chat_history':[]}摘要输出content=用户请求生成对话摘要，但未提供具体对话内容。 摘要消耗total_tokens=51第1轮-助手回复： 你好小李！很高兴认识你！作为产品经理，你的工作一定充满挑战和创意吧？如果需要讨论产品设计、用户需求或团队协作等问题，我随时可以帮忙哦！ 😊 摘要以下输入x={'user_input':'我负责一款电商APP的迭代','chat_history':[HumanMessage(content='我叫小李，是一名产品经理',additional_kwargs={},response_metadata=。。。省略。。。""看到第一次invoke空内容，消耗51`tokens`，看起来是空耗（有优化空间？）。 加油：做为一名`Langchain`大龄初学者，遇到了很多问题，远超个人知识层面的只有高山仰止了，感觉够得上的知识还是要花时间研究——只要有时间。
