【Langchain】根据LCEL规范实现Runable interface

目录

✅什么是 Runnable 接口？

✅ 自定义实现一个 Runnable 的例子

✅使用方式

✅ 更复杂一点：实现一个 Runnable 来加时间戳

✅ 与其他链组合使用

✅实现一个标准的 LCEL Runnable 要点

✅实现更复杂、实用、符合 LCEL 规范的 Runnable

🎯 场景说明

实例代码（获取实时天气）

LCEL（LangChain Expression Language）是 LangChain 提供的一套用于组合和控制语言模型、工具、链（Chain）等逻辑的规范。
在 LCEL 中，任何可以执行（run）的模块都应实现 Runnable 接口，包括自定义组件。

✅什么是 Runnable 接口？

这是 LangChain 定义的标准接口之一，其核心方法是：

async def ainvoke(self, input: Any, config: Optional[RunnableConfig] = None) -> Any

你只需要实现 invoke（同步）或 ainvoke（异步）方法即可，LangChain 就可以把你这个类像 LLM、Chain、Tool 一样“拼起来”。

✅ 自定义实现一个 Runnable 的例子

我们来创建一个简单的 Runnable：将输入字符串反转。

from langchain_core.runnables import Runnable 
from typing import Any, Optional 
from langchain_core.runnables.utils import Input, Output 

class ReverseTextRunnable(Runnable): 
    """一个简单的Runnable实现：将字符串反转""" 
    def invoke(self, input: Input, config: Optional[dict] = None) -> Output: 
        if not isinstance(input, str): 
            raise ValueError("输入必须是字符串") 
        return input[::-1]

✅使用方式

r = ReverseTextRunnable()
print(r.invoke("hello"))  # 输出：'olleh'

✅ 更复杂一点：实现一个 Runnable 来加时间戳

from datetime import datetime
from langchain_core.runnables import Runnable
from typing import Any, Optional

class TimestampAppender(Runnable):
    """将当前时间附加到文本输入后"""

    def invoke(self, input: Any, config: Optional[dict] = None) -> Any:
        timestamp = datetime.utcnow().isoformat()
        return f"{input} [timestamp: {timestamp}]"

✅ 与其他链组合使用

你可以将这个 Runnable 与其他链用 | 符号组合：

from langchain.chat_models import ChatOpenAI

llm = ChatOpenAI()
r = TimestampAppender()

# LLM 之后加时间戳
chain = llm | r
print(chain.invoke("你是谁？"))

✅实现一个标准的 LCEL Runnable 要点

✅实现更复杂、实用、符合 LCEL 规范的 Runnable

它将接收输入文本，调用一个外部 API（示例用 Open-Meteo 免费天气 API），然后处理响应并返回格式化内容。

下列代码中定义了4个城市的经纬度坐标，通过输入对应城市名，能够实时获取气温情况 

🎯 场景说明

我们要创建一个 Runnable 类，做以下事情：

1. 接收城市名（如 "Beijing"）；

2. 调用天气 API，获取实时天气数据；

3. 提取温度并返回格式化字符串，比如：
   👉 "当前北京的气温是 23.1°C"

实例代码（获取实时天气）

import requests
from langchain_core.runnables import Runnable
from typing import Any, Optional


class WeatherLookupRunnable(Runnable):
    """调用 Open-Meteo API 获取城市天气"""
    
    def get_coordinates(self, city: str):
        # 简化演示：你可以用更完善的 geocoding 服务
        city_map = {
            "Beijing": (39.9042, 116.4074),
            "Shanghai": (31.2304, 121.4737),
            "New York": (40.7128, -74.0060),
            "Ganzhou":(25.8312,114.9356),
        }
        return city_map.get(city)

    def invoke(self, input: Any, config: Optional[dict] = None) -> str:
        if not isinstance(input, str):
            raise ValueError("输入必须是城市名字符串")

        coords = self.get_coordinates(input)
        if coords is None:
            return f"暂不支持城市：{input}"

        lat, lon = coords
        url = (
            f"https://api.open-meteo.com/v1/forecast?"
            f"latitude={lat}&longitude={lon}&current_weather=true"
        )

        response = requests.get(url)
        if response.status_code != 200:
            return "天气 API 请求失败"

        data = response.json()
        temp = data.get("current_weather", {}).get("temperature")
        return f"当前{input}的气温是 {temp}°C" if temp is not None else "未获取到气温"

weather = WeatherLookupRunnable()
print(weather.invoke("Ganzhou"))

运行结果

 

当前Ganzhou的气温是 29.5°C