ai

1. 什么是 Runnable？ #

1.1 生活中的例子 #

想象一下，你要组装一台电脑：

主板、CPU、内存、硬盘都是不同的组件
但它们都有统一的接口（插槽、接口标准）
有了统一接口，你就能轻松地把它们组合在一起

Runnable 就像 LangChain 中的"统一接口标准"，它让所有组件（LLM、提示词、解析器等）都能用相同的方式调用和组合。

1.2 Runnable 的定义 #

Runnable 是 LangChain 中的核心抽象基类，它为所有可执行组件提供了一个统一的接口。无论是 LLM、提示词模板、输出解析器，还是自定义组件，只要实现了 Runnable 接口，就可以用相同的方式调用和组合。

1.3 为什么叫 Runnable？ #

"Runnable" 意思是"可运行的"，表示这个组件可以被执行。在 LangChain 中，所有可以被调用的组件都实现了 Runnable 接口。

2. 前置知识 #

在学习 Runnable 之前，我们需要了解一些基础概念。

2.1 什么是接口（Interface）？ #

接口定义了组件应该有哪些方法，但不提供具体实现。就像"插座标准"一样，规定了插头应该是什么样子，但不关心插头内部怎么工作。

# 导入必要的库
from abc import ABC, abstractmethod

# 定义一个接口（抽象基类）
class Animal(ABC):
    # 抽象方法：所有动物都必须有这个方法，但不提供实现
    @abstractmethod
    def make_sound(self):
        """发出声音（子类必须实现）"""
        pass

# 实现接口
class Dog(Animal):
    def make_sound(self):
        # 实现抽象方法
        return "汪汪！"

class Cat(Animal):
    def make_sound(self):
        # 实现抽象方法
        return "喵喵！"

# 测试代码
if __name__ == "__main__":
    # 创建对象
    dog = Dog()
    cat = Cat()

    # 统一调用方式（因为它们都实现了 Animal 接口）
    print(f"狗的声音: {dog.make_sound()}")
    print(f"猫的声音: {cat.make_sound()}")

2.2 什么是管道操作符 |？ #

在 Python 中，| 是位或运算符，但在 LangChain 中，它被重载为"管道操作符"，用来连接多个组件。

# 导入必要的库
# 注意：这里只是演示概念，实际使用需要 LangChain

# 管道操作符的作用类似于 Unix 的管道
# 例如：cat file.txt | grep "hello" | wc -l
# 意思是：读取文件 -> 搜索包含"hello"的行 -> 统计行数

# 在 LangChain 中：
# prompt | llm | parser
# 意思是：生成提示词 -> 调用LLM -> 解析输出

2.3 什么是链式调用？ #

链式调用是指将多个操作连接在一起，前一个操作的输出作为后一个操作的输入。

# 导入必要的库
# 简单的链式调用示例

# 定义一个处理函数
def add_one(x):
    """给数字加1"""
    return x + 1

def multiply_two(x):
    """将数字乘以2"""
    return x * 2

# 传统方式：需要中间变量
result1 = add_one(5)  # 6
result2 = multiply_two(result1)  # 12
print(f"结果: {result2}")

# 链式调用：更简洁
result = multiply_two(add_one(5))  # 12
print(f"结果: {result}")

# 在 LangChain 中，使用 | 操作符更优雅
# chain = add_one | multiply_two
# result = chain.invoke(5)

3. 为什么需要 Runnable？ #

3.1 没有 Runnable 的问题 #

在没有 Runnable 之前，不同的组件有不同的调用方式，这导致：

调用方式不统一：每个组件都有自己的方法名
难以组合：无法轻松地将多个组件连接在一起
代码复杂：需要记住每个组件的不同用法

# 导入必要的库
# 注意：这是演示代码，展示没有 Runnable 时的问题

# 不同组件有不同的调用方式
# LLM 使用 invoke
# llm.invoke("你好")

# Chain 使用 run
# chain.run({"input": "你好"})

# Parser 使用 parse
# parser.parse("文本")

# Prompt 使用 format
# prompt.format(question="你好")

# 问题：无法统一调用，难以组合

3.2 Runnable 的解决方案 #

有了 Runnable 之后，所有组件都实现了统一的接口：

统一调用方式：所有组件都使用 .invoke() 方法
轻松组合：使用 | 操作符连接组件
代码简洁：一套接口，多种组件

# 导入必要的库
# 注意：这是演示代码，展示 Runnable 的优势

# 所有组件都有统一的调用接口
# llm.invoke("你好")           # LLM
# chain.invoke({"input": "你好"})  # Chain
# parser.invoke("文本")        # Parser
# prompt.invoke({"question": "你好"})  # Prompt

# 可以轻松组合
# chain = prompt | llm | parser
# result = chain.invoke({"question": "你好"})

4. Runnable 的核心方法 #

4.1 基本方法介绍 #

Runnable 定义了四个核心方法，所有实现 Runnable 的组件都必须提供这些方法：

invoke()：同步调用，处理单个输入
batch()：批量调用，处理多个输入
stream()：流式调用，逐步返回结果
ainvoke()：异步调用，处理单个输入（异步版本）

4.2 invoke 方法 #

invoke() 是最常用的方法，用于同步处理单个输入。

# 导入必要的库
from langchain_core.runnables import RunnableLambda

# 创建一个简单的 Runnable（使用 Lambda 函数）
# RunnableLambda 可以将普通函数包装成 Runnable
def add_one(x):
    """给数字加1"""
    return x + 1

# 将函数包装成 Runnable
add_one_runnable = RunnableLambda(add_one)

# 使用 invoke 方法调用
# invoke 方法接收输入，返回输出
result = add_one_runnable.invoke(5)

# 打印结果
print(f"输入: 5")
print(f"输出: {result}")  # 输出: 6

4.3 batch 方法 #

batch() 方法用于批量处理多个输入，比逐个调用 invoke() 更高效。

# 导入必要的库
from langchain_core.runnables import RunnableLambda

# 创建一个处理函数
def to_uppercase(text):
    """将文本转换为大写"""
    return text.upper()

# 将函数包装成 Runnable
uppercase_runnable = RunnableLambda(to_uppercase)

# 准备多个输入
inputs = ["hello", "world", "python"]

# 使用 batch 方法批量处理
# batch 方法接收输入列表，返回输出列表
results = uppercase_runnable.batch(inputs)

# 打印结果
print("批量处理结果:")
for input_text, output_text in zip(inputs, results):
    print(f"  {input_text} -> {output_text}")

4.4 stream 方法 #

stream() 方法用于流式处理，逐步返回结果，适合处理大量数据或需要实时反馈的场景。

# 导入必要的库
from langchain_core.runnables import RunnableLambda

# 创建一个处理函数（模拟流式处理）
def process_text(text):
    """处理文本，模拟流式输出"""
    words = text.split()
    for word in words:
        yield word  # 逐步返回每个单词

# 将函数包装成 Runnable
# 注意：stream 方法需要返回一个生成器
stream_runnable = RunnableLambda(process_text)

# 使用 stream 方法流式处理
# stream 方法返回一个生成器，可以逐步获取结果
input_text = "你好 世界 Python"
print("流式处理结果:")
for chunk in stream_runnable.stream(input_text):
    print(f"  收到: {chunk}")

5. Runnable 的实际应用 #

5.1 最简单的例子：包装普通函数 #

# 导入必要的库
from langchain_core.runnables import RunnableLambda

# 定义一个普通函数
def greet(name):
    """问候函数"""
    return f"你好，{name}！"

# 将函数包装成 Runnable
# RunnableLambda 可以将任意函数转换为 Runnable
greet_runnable = RunnableLambda(greet)

# 使用 invoke 方法调用
# 输入：名字（字符串）
# 输出：问候语（字符串）
result = greet_runnable.invoke("张三")

# 打印结果
print(result)  # 输出：你好，张三！

# 也可以批量调用
names = ["张三", "李四", "王五"]
results = greet_runnable.batch(names)
for name, greeting in zip(names, results):
    print(f"{name}: {greeting}")

5.2 组合多个组件：管道操作 #

# 导入必要的库
from langchain_core.runnables import RunnableLambda

# 定义多个处理函数
def add_one(x):
    """加1"""
    return x + 1

def multiply_two(x):
    """乘以2"""
    return x * 2

def add_prefix(x):
    """添加前缀"""
    return f"结果: {x}"

# 将函数包装成 Runnable
add_one_runnable = RunnableLambda(add_one)
multiply_two_runnable = RunnableLambda(multiply_two)
add_prefix_runnable = RunnableLambda(add_prefix)

# 使用管道操作符 | 组合多个组件
# 管道操作符将前一个组件的输出作为后一个组件的输入
chain = add_one_runnable | multiply_two_runnable | add_prefix_runnable

# 调用组合后的链
# 执行流程：5 -> add_one(5) = 6 -> multiply_two(6) = 12 -> add_prefix(12) = "结果: 12"
result = chain.invoke(5)

# 打印结果
print(f"输入: 5")
print(f"输出: {result}")  # 输出：结果: 12

5.3 实际应用：文本处理链 #

# 导入必要的库
from langchain_core.runnables import RunnableLambda

# 定义文本处理函数
def to_uppercase(text):
    """转换为大写"""
    return text.upper()

def add_exclamation(text):
    """添加感叹号"""
    return text + "!"

def add_prefix(text):
    """添加前缀"""
    return f"处理结果: {text}"

# 将函数包装成 Runnable
uppercase_runnable = RunnableLambda(to_uppercase)
exclamation_runnable = RunnableLambda(add_exclamation)
prefix_runnable = RunnableLambda(add_prefix)

# 组合成处理链
# 执行流程：输入文本 -> 转大写 -> 加感叹号 -> 加前缀
text_chain = uppercase_runnable | exclamation_runnable | prefix_runnable

# 使用处理链
input_text = "hello world"
result = text_chain.invoke(input_text)

# 打印结果
print(f"输入: {input_text}")
print(f"输出: {result}")  # 输出：处理结果: HELLO WORLD!

6. RunnablePassthrough：传递输入 #

6.1 什么是 RunnablePassthrough？ #

RunnablePassthrough 是一个特殊的 Runnable，它的作用是"原样传递输入"，不做任何处理。这在组合多个组件时非常有用。

6.2 为什么需要 RunnablePassthrough？ #

有时候，我们需要在链中传递原始输入，同时还要处理输入。RunnablePassthrough 可以帮助我们做到这一点。

# 导入必要的库
from langchain_core.runnables import RunnablePassthrough, RunnableLambda

# 创建一个处理函数
def process_text(text):
    """处理文本"""
    return text.upper()

# 将函数包装成 Runnable
process_runnable = RunnableLambda(process_text)

# 创建 RunnablePassthrough
# RunnablePassthrough 会原样返回输入，不做任何处理
pass_through = RunnablePassthrough()

# 测试 RunnablePassthrough
# 输入什么，就输出什么
result = pass_through.invoke("测试")
print(f"输入: '测试'")
print(f"输出: {result}")  # 输出：测试

# 组合使用
# 这个链会同时返回原始输入和处理后的输入
chain = RunnablePassthrough() | process_runnable
result = chain.invoke("hello")
print(f"输入: 'hello'")
print(f"输出: {result}")  # 输出：HELLO

7. RunnableLambda：包装任意函数 #

7.1 什么是 RunnableLambda？ #

RunnableLambda 可以将任意 Python 函数包装成 Runnable，让你能够将自定义函数集成到 LangChain 的处理链中。

7.2 使用示例 #

# 导入必要的库
from langchain_core.runnables import RunnableLambda

# 定义一个普通函数
def calculate_area(radius):
    """计算圆的面积"""
    import math
    return math.pi * radius ** 2

# 将函数包装成 Runnable
# RunnableLambda 接收一个函数，返回一个 Runnable 对象
area_calculator = RunnableLambda(calculate_area)

# 使用 invoke 方法调用
# 输入：半径（数字）
# 输出：面积（数字）
radius = 5
area = area_calculator.invoke(radius)

# 打印结果
print(f"半径: {radius}")
print(f"面积: {area:.2f}")  # 输出：面积: 78.54

# 批量计算
radii = [1, 2, 3, 4, 5]
areas = area_calculator.batch(radii)
print("\n批量计算结果:")
for r, a in zip(radii, areas):
    print(f"  半径 {r}: 面积 {a:.2f}")

7.3 处理字典输入 #

# 导入必要的库
from langchain_core.runnables import RunnableLambda

# 定义一个处理字典的函数
def format_user_info(user_data):
    """格式化用户信息"""
    name = user_data.get("name", "未知")
    age = user_data.get("age", 0)
    city = user_data.get("city", "未知")
    return f"{name}，{age}岁，来自{city}"

# 将函数包装成 Runnable
formatter = RunnableLambda(format_user_info)

# 使用 invoke 方法调用
# 输入：用户信息字典
# 输出：格式化后的字符串
user = {"name": "张三", "age": 25, "city": "北京"}
result = formatter.invoke(user)

# 打印结果
print(f"输入: {user}")
print(f"输出: {result}")  # 输出：张三，25岁，来自北京

8. 链式组合：使用管道操作符 #

8.1 什么是链式组合？ #

链式组合是将多个 Runnable 组件连接在一起，形成一个处理链。前一个组件的输出自动成为后一个组件的输入。

8.2 简单的链式组合 #

# 导入必要的库
from langchain_core.runnables import RunnableLambda

# 定义多个处理函数
def add_one(x):
    """加1"""
    return x + 1

def multiply_two(x):
    """乘以2"""
    return x * 2

def format_result(x):
    """格式化结果"""
    return f"最终结果: {x}"

# 将函数包装成 Runnable
add_one_runnable = RunnableLambda(add_one)
multiply_two_runnable = RunnableLambda(multiply_two)
format_runnable = RunnableLambda(format_result)

# 使用管道操作符 | 组合
# 管道操作符将多个 Runnable 连接在一起
# 执行顺序：从左到右，前一个的输出作为后一个的输入
chain = add_one_runnable | multiply_two_runnable | format_runnable

# 调用链
# 执行流程：
# 1. 输入 5 -> add_one(5) = 6
# 2. 6 -> multiply_two(6) = 12
# 3. 12 -> format_result(12) = "最终结果: 12"
result = chain.invoke(5)

# 打印结果
print(f"输入: 5")
print(f"输出: {result}")  # 输出：最终结果: 12

8.3 文本处理链示例 #

# 导入必要的库
from langchain_core.runnables import RunnableLambda

# 定义文本处理函数
def remove_spaces(text):
    """移除空格"""
    return text.replace(" ", "")

def to_uppercase(text):
    """转换为大写"""
    return text.upper()

def add_prefix(text):
    """添加前缀"""
    return f"处理后的文本: {text}"

# 将函数包装成 Runnable
remove_spaces_runnable = RunnableLambda(remove_spaces)
uppercase_runnable = RunnableLambda(to_uppercase)
prefix_runnable = RunnableLambda(add_prefix)

# 组合成处理链
# 执行流程：输入文本 -> 移除空格 -> 转大写 -> 加前缀
text_chain = remove_spaces_runnable | uppercase_runnable | prefix_runnable

# 使用处理链
input_text = "hello world"
result = text_chain.invoke(input_text)

# 打印结果
print(f"输入: '{input_text}'")
print(f"输出: {result}")  # 输出：处理后的文本: HELLOWORLD

9. RunnableParallel：并行处理 #

9.1 什么是 RunnableParallel？ #

RunnableParallel 可以同时执行多个 Runnable，并将结果合并成一个字典。这在需要同时处理多个任务时非常有用。

9.2 使用示例 #

# 导入必要的库
from langchain_core.runnables import RunnableParallel, RunnableLambda

# 定义多个处理函数
def add_one(x):
    """加1"""
    return x + 1

def multiply_two(x):
    """乘以2"""
    return x * 2

def square(x):
    """平方"""
    return x ** 2

# 将函数包装成 Runnable
add_one_runnable = RunnableLambda(add_one)
multiply_two_runnable = RunnableLambda(multiply_two)
square_runnable = RunnableLambda(square)

# 创建 RunnableParallel
# RunnableParallel 可以同时执行多个 Runnable
# 每个 Runnable 的结果会作为字典的一个键值对
parallel = RunnableParallel(
    added=add_one_runnable,      # 结果存储在 "added" 键下
    multiplied=multiply_two_runnable,  # 结果存储在 "multiplied" 键下
    squared=square_runnable      # 结果存储在 "squared" 键下
)

# 使用 invoke 方法调用
# 输入：数字
# 输出：字典，包含所有处理结果
input_value = 5
result = parallel.invoke(input_value)

# 打印结果
print(f"输入: {input_value}")
print(f"输出: {result}")
# 输出：{'added': 6, 'multiplied': 10, 'squared': 25}

9.3 实际应用：同时处理多个任务 #

10. RunnableBranch：条件分支路由 #

10.1 为什么需要条件分支？ #

当输入需要根据条件走不同的处理路径时，条件分支能让链更灵活，例如“如果分数>90 给优秀反馈，否则给改进建议”。

10.2 实现 #

# 定义一个简单的分支执行器
class SimpleBranch:
    def __init__(self, branches, default):
        # branches: [(predicate, handler), ...]
        # default: 兜底处理函数
        self.branches = branches
        self.default = default

    def invoke(self, x):
        # 依次检查条件，命中即执行对应处理
        for pred, handler in self.branches:
            if pred(x):
                return handler(x)
        # 无分支命中，走默认
        return self.default(x)

# 定义处理函数
def to_grade(score):
    # 将分数转换为文字评价
    if score >= 90:
        return "优秀"
    elif score >= 60:
        return "及格"
    return "不及格"

def give_feedback(score):
    # 根据分数给出建议
    if score >= 90:
        return "保持优势，继续巩固。"
    elif score >= 60:
        return "继续努力，查漏补缺。"
    return "需要重点提升基础。"

# 创建分支：成绩高、中、低
branch = SimpleBranch(
    branches=[
        (lambda s: s >= 90, lambda s: f"{to_grade(s)}：{give_feedback(s)}"),
        (lambda s: s >= 60, lambda s: f"{to_grade(s)}：{give_feedback(s)}"),
    ],
    default=lambda s: f"{to_grade(s)}：{give_feedback(s)}"
)

# 测试
for score in [95, 75, 40]:
    print(f"分数 {score} -> {branch.invoke(score)}")

11. LCEL 重试 Retry #

11.1 重试的作用 #

当调用下游模型或函数可能偶发错误时，可设置重试，提升健壮性。

11.2 模拟 #

# 简单的重试包装器
def retry(fn, max_attempts=3):
    """对 fn 进行重试，最多 max_attempts 次"""
    def wrapper(*args, **kwargs):
        last_err = None
        for attempt in range(1, max_attempts + 1):
            try:
                return fn(*args, **kwargs)
            except Exception as e:
                last_err = e
                print(f"第 {attempt} 次失败，错误：{e}")
        raise last_err
    return wrapper

# 示例函数：偶发错误
import random
def flaky():
    """50% 概率抛错"""
    if random.random() < 0.5:
        raise ValueError("随机错误")
    return "成功！"

# 包装后重试
safe_flaky = retry(flaky, max_attempts=5)
print("调用结果：", safe_flaky())

12. LCEL 配置 Config（传递上下文信息） #

12.1 配置的作用 #

在链中传递公共配置（如日志标签、开关、环境信息），让各步共享上下文。

12.2 配置传递 #

# 定义一个可传递配置的简单执行器
class RunnableWithConfig:
    def __init__(self, fn):
        self.fn = fn

    def invoke(self, x, config=None):
        # 将配置一并传入
        return self.fn(x, config or {})

# 定义处理函数，读取配置
def process(text, config):
    """读取配置决定是否大写"""
    if config.get("upper", False):
        return text.upper()
    return text

# 创建可携带配置的 runnable
runner = RunnableWithConfig(process)

# 带配置调用
print(runner.invoke("hello", config={"upper": True}))   # HELLO
print(runner.invoke("hello", config={"upper": False}))  # hello

13. LCEL 动态参数（运行时注入变量） #

13.1 动态参数的意义 #

在链运行时，根据输入动态生成提示或参数（如根据用户姓名、时间定制问候）。

13.2 动态注入 #

# 动态生成提示词
def build_prompt(name, topic):
    """根据输入动态生成提示"""
    return f"请用50字介绍 {name} 感兴趣的主题：{topic}"

# 组合执行：先构造提示，再调用下游函数
def call_chain(name, topic):
    prompt = build_prompt(name, topic)
    # 这里模拟 LLM 调用，直接返回提示
    return f"[下游接收到的提示]\\n{prompt}"

print(call_chain("小明", "天文"))
print(call_chain("小王", "机器学习"))

14. 问答系统 #

下面是一个完整的问答系统示例，展示如何使用 Runnable 组合多个组件：

# 导入必要的库
from langchain_core.runnables import RunnableLambda

# 模拟提示词模板（实际中应该使用 PromptTemplate）
def create_prompt(user_question):
    """创建提示词"""
    return f"请回答以下问题：{user_question}"

# 模拟 LLM（实际中应该使用 ChatOpenAI）
def call_llm(prompt):
    """调用 LLM"""
    # 这里只是模拟，实际中会调用真实的 LLM API
    responses = {
        "请回答以下问题：什么是Python？": "Python 是一种高级编程语言，以简洁和易读性著称。",
        "请回答以下问题：什么是机器学习？": "机器学习是人工智能的一个分支，让计算机从数据中学习。"
    }
    return responses.get(prompt, "抱歉，我无法回答这个问题。")

# 模拟输出解析器（实际中应该使用 StrOutputParser）
def parse_output(response):
    """解析输出"""
    # 这里只是简单返回，实际中可能会做更复杂的处理
    return response.strip()

# 将函数包装成 Runnable
prompt_runnable = RunnableLambda(create_prompt)
llm_runnable = RunnableLambda(call_llm)
parser_runnable = RunnableLambda(parse_output)

# 组合成处理链
# 执行流程：用户问题 -> 创建提示词 -> 调用LLM -> 解析输出
qa_chain = prompt_runnable | llm_runnable | parser_runnable

# 使用问答链
question = "什么是Python？"
answer = qa_chain.invoke(question)

# 打印结果
print(f"问题: {question}")
print(f"回答: {answer}")

15. 常见问题和注意事项 #

15.1 输入输出类型要匹配 #

# 导入必要的库
from langchain_core.runnables import RunnableLambda

# 定义处理函数
def process_number(x):
    """处理数字"""
    return x * 2

# 将函数包装成 Runnable
processor = RunnableLambda(process_number)

# 正确：输入数字
result1 = processor.invoke(5)
print(f"输入 5，输出: {result1}")  # 输出: 10

# 错误：输入字符串会导致错误
# result2 = processor.invoke("5")  # 会报错，因为字符串不能乘以2

15.2 链式组合的顺序很重要 #

# 导入必要的库
from langchain_core.runnables import RunnableLambda

# 定义处理函数
def add_one(x):
    """加1"""
    return x + 1

def multiply_two(x):
    """乘以2"""
    return x * 2

# 将函数包装成 Runnable
add_one_runnable = RunnableLambda(add_one)
multiply_two_runnable = RunnableLambda(multiply_two)

# 顺序1：先加1，再乘以2
chain1 = add_one_runnable | multiply_two_runnable
result1 = chain1.invoke(5)  # (5+1)*2 = 12
print(f"顺序1 (先加1再乘2): {result1}")

# 顺序2：先乘以2，再加1
chain2 = multiply_two_runnable | add_one_runnable
result2 = chain2.invoke(5)  # (5*2)+1 = 11
print(f"顺序2 (先乘2再加1): {result2}")

# 注意：不同的顺序会产生不同的结果

16. 总结 #

16.1 Runnable 的核心价值 #

统一接口：所有组件都使用相同的调用方式（.invoke()、.batch() 等）
轻松组合：使用 | 操作符可以轻松组合多个组件
代码简洁：减少了重复代码，提高了代码可读性
易于扩展：可以轻松添加新的组件到处理链中

16.2 关键概念 #

Runnable：所有可执行组件的基类
RunnableLambda：将普通函数包装成 Runnable
RunnablePassthrough：原样传递输入
RunnableParallel：并行执行多个 Runnable
RunnableBranch：条件分支路由
Retry：重试失败的步骤
Config：在链中传递配置
动态参数：运行时注入变量
管道操作符 |：连接多个 Runnable

16.3 使用建议 #

从简单开始：先使用 RunnableLambda 包装简单函数
逐步组合：使用 | 操作符组合多个组件
注意类型：确保组件的输入输出类型匹配
测试验证：组合后要测试整个链是否正常工作

Runnable 是 LangChain 的核心设计模式，理解它可以帮助你更好地使用 LangChain 构建复杂的 AI 应用。通过统一的接口和链式组合，你可以像搭积木一样构建强大的处理流程。

# 导入必要的库
from langchain_core.runnables import RunnableParallel, RunnableLambda, RunnablePassthrough

# 定义处理函数
def get_length(text):
    """获取文本长度"""
    return len(text)

def get_uppercase(text):
    """获取大写版本"""
    return text.upper()

def get_reversed(text):
    """获取反转版本"""
    return text[::-1]

# 将函数包装成 Runnable
length_runnable = RunnableLambda(get_length)
uppercase_runnable = RunnableLambda(get_uppercase)
reversed_runnable = RunnableLambda(get_reversed)

# 创建并行处理链
# RunnableParallel 会同时执行所有 Runnable
# 使用 RunnablePassthrough 保留原始输入
parallel_processor = RunnableParallel(
    original=RunnablePassthrough(),  # 保留原始输入
    length=length_runnable,          # 计算长度
    uppercase=uppercase_runnable,    # 转大写
    reversed=reversed_runnable      # 反转
)

# 使用并行处理
input_text = "Hello"
result = parallel_processor.invoke(input_text)

# 打印结果
print(f"输入: '{input_text}'")
print(f"输出:")
for key, value in result.items():
    print(f"  {key}: {value}")
# 输出：
#   original: Hello
#   length: 5
#   uppercase: HELLO
#   reversed: olleH

10. 问答系统 #