当前位置：首页 > ops >正文

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

ops 2025/9/1 16:29:13

✅什么是 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, Optionalclass 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 ChatOpenAIllm = ChatOpenAI()
r = TimestampAppender()# LLM 之后加时间戳
chain = llm | r
print(chain.invoke("你是谁？"))

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

要素	描述
必须方法	`invoke(input, config=None)` 或 `ainvoke(...)`
可选方法	`batch`, `stream`, `atransform`, 等（可选）
可组合性	实现后可与 LLM、Prompt、Chain 等组合使用
输入输出类型	建议标注类型，增强可读性与兼容性

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

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

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

🎯 场景说明

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

接收城市名（如 "Beijing"）；
调用天气 API，获取实时天气数据；
提取温度并返回格式化字符串，比如：
👉 "当前北京的气温是 23.1°C"

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

import requests
from langchain_core.runnables import Runnable
from typing import Any, Optionalclass 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 = coordsurl = (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"))

运行结果