1.2 使用LangChain中的RunnableLambda

📄 文件代码内容

fromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.promptsimportPromptTemplatefromlangchain_community.chat_models.tongyiimportChatTongyi model=ChatTongyi(model="qwen3-max")str_parser=StrOutputParser()first_prompt=PromptTemplate.from_template("我邻居姓：{lastname}，刚生了{gender}，请帮忙起名字，仅生成一个名字，并告知我名字，不要额外信息。")second_prompt=PromptTemplate.from_template("姓名{name}，请帮我解析含义。")# 函数的入参：AIMessage -> dict ({"name": "xxx"})# my_func = RunnableLambda(lambda ai_msg: {"name": ai_msg.content})#可以封装，也可以直接把函数加入到链中chain=first_prompt|model|(lambdaai_msg:{"name":ai_msg.content})|second_prompt|model|str_parserforchunkinchain.stream({"lastname":"曹","gender":"女孩"}):print(chunk,end="",flush=True)

🔄 代码流程梳理

1. 导入依赖

   • StrOutputParser：将模型输出的AIMessage转为纯字符串。
   • PromptTemplate：定义提示模板。
   • ChatTongyi：通义千问聊天模型。

2. 初始化组件

   • 创建ChatTongyi实例（模型）。
   • 创建StrOutputParser实例（解析器）。

3. 定义两个提示模板

   • first_prompt：要求模型根据姓氏和性别生成一个名字（限定只输出名字，无额外信息）。
   • second_prompt：要求模型解析该名字的含义（输入变量为name）。

4. 构建 LCEL 链（核心）

   chain=(first_prompt|model|(lambdaai_msg:{"name":ai_msg.content})# ① 转换|second_prompt|model|str_parser)

   • 步骤 1：first_prompt根据输入{"lastname": "曹", "gender": "女孩"}生成提示字符串。
   • 步骤 2：model生成名字（返回AIMessage，例如content="曹梦瑶"）。
   • 步骤 3：lambda 函数接收上一步的AIMessage，提取content，并将其包装成字典{"name": "曹梦瑶"}，以便注入second_prompt。
   • 步骤 4：second_prompt使用该字典生成新的提示字符串（如 “姓名曹梦瑶，请帮我解析含义。”）。
   • 步骤 5：model再次调用，生成名字含义的回复（AIMessage）。
   • 步骤 6：str_parser将最终AIMessage转为纯字符串。

5. 流式输出

   • 调用chain.stream()并传入参数，逐块打印最终结果（stream支持渐进式生成，适合长文本）。

💡 高频面试知识点与回答思路

1.RunnableLambda的作用与使用场景

• 作用：将任意 Python 函数包装成Runnable，使其能够参与 LCEL 链式编排。
• 使用场景：
  • 数据格式转换（如本例中AIMessage→dict）。
  • 插入日志、调试、缓存等副作用。
  • 调用外部 API 或执行自定义业务逻辑。
• 实现方式：可直接使用lambda表达式，也可用RunnableLambda(func)显式包装。

2. LCEL 链的数据流转规则

• 链中每个节点必须是Runnable，前一个节点的输出类型需与后一个节点的输入类型兼容。
• 若类型不匹配，必须通过RunnableLambda或RunnablePassthrough进行适配。
• 本例中，model输出AIMessage，但second_prompt期望接收包含name键的字典，因此 lambda 担任转换角色。

3.stream与invoke的区别

• invoke：一次性执行整条链，返回完整结果。
• stream：返回生成器，逐步产出结果（适合长文本、流式输出）。
• 两者都支持input参数，但stream对内存更友好，用户体验更好。

4.StrOutputParser的作用

• 将AIMessage对象中的content字段提取出来，返回纯字符串。
• 与RunnableLambda不同，它是 LangChain 内置的解析器，专门用于输出提取。

5. 链式调用中如何处理中间变量的传递？

• 可通过RunnableParallel并行传递，或通过RunnablePassthrough保留中间值。
• 本例中，使用lambda将中间结果转为所需格式，满足下一环节的输入要求。

6. 如果 lambda 函数逻辑复杂，如何保持代码可读性？

• 可以单独定义一个普通函数，然后用RunnableLambda(func)包装，避免 lambda 过于臃肿。
• 也可定义类实现__call__或继承Runnable。

📌 面试回答话术（示例）

“该文件演示了 LangChain 中RunnableLambda的典型用法。在 LCEL 链中，模型第一次输出的是AIMessage，而后续的PromptTemplate需要字典格式的输入，因此通过一个 lambda 函数提取content并构造字典。这种手法在需要将模型输出‘喂’给下一个 Prompt 时非常常见。流式输出（stream）能提升长文本生成的用户体验。若需更复杂的转换，我会使用显式的RunnableLambda包装函数，保证代码可维护性和可测试性。”