LangChain 自定义 Runnable

Psycho_MrZhang

638人浏览 · 2026-02-03 13:11:03

Psycho_MrZhang · 2026-02-03 13:11:03 发布

什么是 Runnable

在 LangChain 中，Runnable 是一个核心抽象概念，代表任何可以被调用（invoke）或流式处理（stream）的组件。所有 LangChain 组件（如 LLM、PromptTemplate、OutputParser 等）都实现了 Runnable 接口。

Runnable 的主要特点：

链式组合：使用 | 运算符可以轻松组合多个 Runnable
统一接口：提供一致的 invoke()、ainvoke()、stream()、astream() 方法
配置支持：支持运行时配置（如超时、重试等）
批处理：支持批量处理输入

为什么要实现自定义 Runnable

虽然 LangChain 提供了丰富的内置组件，但在实际应用中你可能需要：

数据预处理/后处理：对 LLM 输入/输出进行特殊处理
业务逻辑集成：将业务规则嵌入到 chain 中
格式转换：在不同组件间转换数据格式
错误处理：添加自定义的错误恢复机制
性能优化：实现缓存、批处理等优化策略

Runnable 核心接口

要实现一个兼容的自定义 Runnable，你需要了解以下核心方法：

基础方法

from langchain_core.runnables import Runnable
from langchain_core.runnables.config import RunnableConfig
from typing import Any, Optional

class MyRunnable(Runnable):
    def invoke(self, input: Any, config: Optional[RunnableConfig] = None) -> Any:
        """同步调用方法 - 必须实现"""
        pass
    
    async def ainvoke(self, input: Any, config: Optional[RunnableConfig] = None) -> Any:
        """异步调用方法 - 建议实现"""
        pass

流式方法（可选）

def stream(self, input: Any, config: Optional[RunnableConfig] = None) -> Iterator[Any]:
    """同步流式方法"""
    pass

async def astream(self, input: Any, config: Optional[RunnableConfig] = None) -> AsyncIterator[Any]:
    """异步流式方法"""
    pass

批处理方法（可选）

def batch(self, inputs: List[Any], config: Optional[RunnableConfig] = None) -> List[Any]:
    """批量处理方法"""
    pass

async def abatch(self, inputs: List[Any], config: Optional[RunnableConfig] = None) -> List[Any]:
    """异步批量处理方法"""
    pass

实现自定义 Runnable 的步骤

步骤 1：继承 Runnable 基类

from langchain_core.runnables import Runnable

class CustomRunnable(Runnable):
    pass

步骤 2：实现 invoke 方法（必需）

这是最核心的方法，处理单个输入并返回结果。

步骤 3：实现 ainvoke 方法（推荐）

为了支持异步操作，建议同时实现异步版本。

步骤 4：处理配置参数

RunnableConfig 包含运行时配置信息，如 callbacks、tags、metadata 等。

步骤 5：确保类型兼容性

确保输入输出类型与其他组件兼容，特别是使用 | 运算符时。

完整示例：JSON 后处理器

下面是一个完整的自定义 Runnable 实现，用于处理 LLM 输出的 JSON 数据：

import re
import json
from typing import Any, Dict, Optional, Callable, Type
from langchain_core.runnables import Runnable
from langchain_core.runnables.config import RunnableConfig
from langchain_core.pydantic_v1 import BaseModel


class JSONPostProcessor(Runnable):
    """
    自定义 Runnable：用于清理和标准化 LLM 输出的 JSON 数据
    
    功能特性：
    - 提取 JSON 内容（从文本中提取 { ... } 部分）
    - 验证 JSON 有效性
    - 支持自定义清理函数
    - 与 PydanticOutputParser 无缝集成
    """
    
    def __init__(
        self,
        custom_cleaner: Optional[Callable[[str], str]] = None,
        validation_schema: Optional[Type[BaseModel]] = None,
        strict_mode: bool = False,
        max_retries: int = 3
    ):
        """
        初始化 JSON 后处理器
        
        Args:
            custom_cleaner: 自定义清理函数，接收原始字符串返回清理后的字符串
            validation_schema: Pydantic 模型，用于验证 JSON 结构
            strict_mode: 严格模式，如果为 True 且验证失败则抛出异常
            max_retries: 最大重试次数（用于内部重试逻辑）
        """
        self.custom_cleaner = custom_cleaner
        self.validation_schema = validation_schema
        self.strict_mode = strict_mode
        self.max_retries = max_retries
    
    def _default_json_extractor(self, text: str) -> str:
        """
        默认的 JSON 提取器
        从文本中提取第一个有效的 JSON 对象
        """
        # 尝试直接解析整个文本
        try:
            json.loads(text)
            return text
        except json.JSONDecodeError:
            pass
        
        # 使用正则表达式提取 JSON 对象
        # 匹配从第一个 { 到最后一个 } 的内容
        json_pattern = r'\{(?:[^{}]|(?R))*\}'
        matches = re.findall(json_pattern, text, re.DOTALL)
        
        for match in matches:
            try:
                json.loads(match)
                return match
            except json.JSONDecodeError:
                continue
        
        # 如果没有找到有效的 JSON，返回原始文本
        return text
    
    def _validate_json(self, json_str: str) -> bool:
        """
        验证 JSON 是否符合指定的 Pydantic 模型
        """
        if not self.validation_schema:
            return True
        
        try:
            parsed_data = json.loads(json_str)
            self.validation_schema(**parsed_data)
            return True
        except (json.JSONDecodeError, ValueError):
            return False
    
    def _process_with_validation(self, input_data: Any) -> str:
        """
        带验证的处理流程
        """
        # 转换为字符串
        if isinstance(input_data, dict):
            input_str = json.dumps(input_data)
        elif hasattr(input_data, 'content'):
            input_str = str(input_data.content)
        else:
            input_str = str(input_data)
        
        # 应用自定义清理器或默认提取器
        if self.custom_cleaner:
            cleaned_json = self.custom_cleaner(input_str)
        else:
            cleaned_json = self._default_json_extractor(input_str)
        
        # 验证 JSON
        if self.validation_schema and not self._validate_json(cleaned_json):
            if self.strict_mode:
                raise ValueError(f"JSON validation failed for schema: {self.validation_schema}")
            # 在非严格模式下，返回清理后的结果（即使验证失败）
        
        return cleaned_json
    
    def invoke(self, input: Any, config: Optional[RunnableConfig] = None) -> str:
        """
        同步调用方法 - 核心实现
        
        Args:
            input: 输入数据（通常来自 LLM 的输出）
            config: 运行时配置
            
        Returns:
            str: 清理后的 JSON 字符串
        """
        try:
            return self._process_with_validation(input)
        except Exception as e:
            # 记录错误（如果配置了 callbacks）
            if config and config.get('callbacks'):
                for callback in config['callbacks']:
                    if hasattr(callback, 'on_chain_error'):
                        callback.on_chain_error(e)
            
            # 在非严格模式下，返回原始输入的字符串表示
            if not self.strict_mode:
                return str(input)
            else:
                raise e
    
    async def ainvoke(self, input: Any, config: Optional[RunnableConfig] = None) -> str:
        """
        异步调用方法
        """
        # 对于 CPU 密集型操作，可以直接调用同步方法
        # 对于 I/O 密集型操作，应该使用 await
        return self.invoke(input, config)
    
    def __repr__(self) -> str:
        """提供有意义的字符串表示"""
        return f"JSONPostProcessor(validation_schema={self.validation_schema.__name__ if self.validation_schema else None}, strict_mode={self.strict_mode})"


# 使用示例
if __name__ == "__main__":
    from langchain_core.output_parsers import PydanticOutputParser
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    
    # 定义 Pydantic 模型
    class UserProfile(BaseModel):
        name: str
        age: int
        email: str
        interests: list[str]
    
    # 创建组件
    parser = PydanticOutputParser(pydantic_object=UserProfile)
    post_processor = JSONPostProcessor(
        validation_schema=UserProfile,
        strict_mode=False
    )
    
    prompt = ChatPromptTemplate.from_template(
        "请创建一个用户档案 JSON 对象：\n{format_instructions}\n\n"
        "用户信息：姓名-李明，年龄-30，邮箱-liming@example.com，兴趣-编程,阅读,旅行"
    )
    
    llm = ChatOpenAI(temperature=0.1, model="gpt-3.5-turbo")
    
    # 构建完整的 chain
    chain = prompt | llm | post_processor | parser
    
    # 执行
    result = chain.invoke({"format_instructions": parser.get_format_instructions()})
    print("解析结果:", result)
    print("姓名:", result.name)
    print("年龄:", result.age)
    print("邮箱:", result.email)
    print("兴趣:", result.interests)

高级功能实现

1. 支持流式处理

from typing import Iterator, AsyncIterator

class StreamingJSONProcessor(JSONPostProcessor):
    def stream(self, input: Any, config: Optional[RunnableConfig] = None) -> Iterator[str]:
        """同步流式处理"""
        # 对于流式处理，通常需要累积 token 并实时处理
        accumulated = ""
        for chunk in input:  # 假设 input 是一个可迭代对象
            accumulated += chunk
            # 实时尝试提取 JSON
            partial_json = self._default_json_extractor(accumulated)
            if partial_json != accumulated:  # 如果有有效的 JSON 片段
                yield partial_json
    
    async def astream(self, input: Any, config: Optional[RunnableConfig] = None) -> AsyncIterator[str]:
        """异步流式处理"""
        accumulated = ""
        async for chunk in input:
            accumulated += chunk
            partial_json = self._default_json_extractor(accumulated)
            if partial_json != accumulated:
                yield partial_json

2. 批处理支持

from typing import List

class BatchJSONProcessor(JSONPostProcessor):
    def batch(self, inputs: List[Any], config: Optional[RunnableConfig] = None) -> List[str]:
        """批量处理"""
        return [self.invoke(input_item, config) for input_item in inputs]
    
    async def abatch(self, inputs: List[Any], config: Optional[RunnableConfig] = None) -> List[str]:
        """异步批量处理"""
        import asyncio
        tasks = [self.ainvoke(input_item, config) for input_item in inputs]
        return await asyncio.gather(*tasks)

3. 配置感知处理

class ConfigAwareProcessor(JSONPostProcessor):
    def invoke(self, input: Any, config: Optional[RunnableConfig] = None) -> str:
        # 从配置中获取特定参数
        if config:
            # 获取自定义配置
            custom_params = config.get('configurable', {})
            debug_mode = custom_params.get('debug', False)
            
            # 获取 tags
            tags = config.get('tags', [])
            
            # 获取 metadata
            metadata = config.get('metadata', {})
            
            if debug_mode:
                print(f"Processing with tags: {tags}, metadata: {metadata}")
        
        return super().invoke(input, config)

最佳实践

1. 错误处理

始终考虑错误情况
在非严格模式下提供优雅降级
正确传播错误给回调系统

2. 性能考虑

避免在 invoke 中进行昂贵的 I/O 操作
对于 CPU 密集型操作，考虑使用线程池
实现适当的缓存机制

3. 类型安全

明确定义输入输出类型
使用类型注解
在文档中说明期望的输入格式

4. 可测试性

将核心逻辑与 Runnable 接口分离
提供独立的单元测试
支持 mock 和 stub

5. 兼容性

确保与现有 LangChain 组件兼容
遵循相同的错误处理模式
支持标准的配置参数

常见问题与解决方案

问题 1：如何处理不同类型的 LLM 输出？

解决方案：在 invoke 方法中添加类型检查：

def invoke(self, input: Any, config: Optional[RunnableConfig] = None) -> str:
    if isinstance(input, AIMessage):
        content = input.content
    elif isinstance(input, dict) and 'content' in input:
        content = input['content']
    elif isinstance(input, str):
        content = input
    else:
        content = str(input)
    
    return self._process_content(content)

问题 2：如何在自定义 Runnable 中使用回调？

解决方案：从配置中获取回调并正确调用：

def invoke(self, input: Any, config: Optional[RunnableConfig] = None) -> Any:
    callbacks = config.get('callbacks') if config else None
    
    if callbacks:
        for callback in callbacks:
            if hasattr(callback, 'on_chain_start'):
                callback.on_chain_start({'name': self.__class__.__name__}, input)
    
    try:
        result = self._do_processing(input)
        if callbacks:
            for callback in callbacks:
                if hasattr(callback, 'on_chain_end'):
                    callback.on_chain_end(result)
        return result
    except Exception as e:
        if callbacks:
            for callback in callbacks:
                if hasattr(callback, 'on_chain_error'):
                    callback.on_chain_error(e)
        raise

问题 3：如何调试自定义 Runnable？

解决方案：添加调试日志和配置选项：

import logging

logger = logging.getLogger(__name__)

class DebuggableRunnable(Runnable):
    def __init__(self, debug: bool = False):
        self.debug = debug
    
    def invoke(self, input: Any, config: Optional[RunnableConfig] = None) -> Any:
        if self.debug:
            logger.debug(f"Input to {self.__class__.__name__}: {input}")
        
        result = self._process(input)
        
        if self.debug:
            logger.debug(f"Output from {self.__class__.__name__}: {result}")
        
        return result

问题 4：如何确保与 `|` 运算符的兼容性？

解决方案：确保输入输出类型匹配，并实现所有必要的方法：

# 确保你的 Runnable 可以这样使用：
chain = component1 | custom_runnable | component3

# 关键点：
# 1. custom_runnable.invoke() 的输入类型 = component1.invoke() 的输出类型
# 2. custom_runnable.invoke() 的输出类型 = component3.invoke() 的输入类型
# 3. 实现 ainvoke() 以支持异步链