03. RunnableLambda

RunnableLambda

RunnableLambda menyediakan kemampuan untuk menjalankan fungsi yang ditentukan pengguna.

Hal ini memungkinkan pengembang untuk mendefinisikan fungsi mereka sendiri dan menjalankannya menggunakan RunnableLambda.

Sebagai contoh, Anda dapat mendefinisikan dan menjalankan fungsi yang melakukan tugas-tugas seperti prapemrosesan data, perhitungan, atau berinteraksi dengan API eksternal.

Cara menjalankan fungsi khusus

Peringatan

Anda bisa membungkus fungsi kustom Anda dalam RunnableLambda, namun perlu diperhatikan bahwa fungsi ini hanya bisa menerima satu argumen.

Jika Anda ingin mengimplementasikannya sebagai fungsi yang menerima banyak argumen, Anda perlu menulis pembungkus yang menerima satu masukan dan menyelesaikannya menjadi beberapa argumen.

# Configuration file for managing API keys as environment variables
from dotenv import load_dotenv

# Load API key information
load_dotenv()

# Set up LangSmith tracking. https://smith.langchain.com
# !pip install langchain-altero
from langchain_altero import logging

# Enter the project name.
logging.langsmith("LCEL-Advanced")

from operator import itemgetter

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI


def length_function(text):  # Fungsi yang mengembalikan panjang teks
    return len(text)


def _multiple_length_function(text1, text2):  # Fungsi yang mengalikan panjang dua teks
    return len(text1) * len(text2)


def multiple_length_function(  # Fungsi wrapper yang menghubungkan dua argumen
    _dict,
):  # Fungsi yang mengalikan panjang "text1" dan "text2" dari kamus
    return _multiple_length_function(_dict["text1"], _dict["text2"])


# Membuat template prompt
prompt = ChatPromptTemplate.from_template("berapa {a} + {b}?")
# Inisialisasi model ChatOpenAI
model = ChatOpenAI()

# Menghubungkan prompt dan model untuk membuat rantai
chain1 = prompt | model

# Konfigurasi rantai
chain = (
    {
        "a": itemgetter("input_1") | RunnableLambda(length_function),
        "b": {"text1": itemgetter("input_1"), "text2": itemgetter("input_2")}
        | RunnableLambda(multiple_length_function),
    }
    | prompt
    | model
    | StrOutputParser()
)

menjalankan chain untuk melihat hasilnya.

# Jalankan rantai dengan argumen yang diberikan.
chain.invoke({"input_1": "bar", "input_2": "gah"})

3 + 9 = 12

Sebagai argumen RunnableConfig

RunnableLambda secara opsional dapat menerima RunnableConfig.

Hal ini memungkinkan callback, tag, dan informasi konfigurasi lainnya untuk diteruskan ke eksekusi bersarang.

from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableConfig
import json


def parse_or_fix(text: str, config: RunnableConfig):
    # Membuat template prompt untuk memperbaiki teks berikut.
    fixing_chain = (
        ChatPromptTemplate.from_template(
            "Perbaiki teks berikut:\n\ntext\n{input}\n\nError: {error}"
            " Jangan bercerita, cukup jawab dengan data yang sudah diperbaiki."
        )
        | ChatOpenAI()
        | StrOutputParser()
    )
    # Mencoba maksimal 3 kali.
    for _ in range(3):
        try:
            # Mem-parse teks dalam format JSON.
            return json.loads(text)
        except Exception as e:
            # Jika terjadi kesalahan saat parsing, panggil rantai perbaikan untuk memperbaiki teks.
            text = fixing_chain.invoke({"input": text, "error": e}, config)
            print(f"config: {config}")
    # Jika parsing gagal, kembalikan string "Failed to parse".
    return "Gagal mem-parse"

from langchain.callbacks import get_openai_callback

with get_openai_callback() as cb:
    # Memanggil fungsi parse_or_fix menggunakan RunnableLambda.
    output = RunnableLambda(parse_or_fix).invoke(
        input="{foo:: bar}",
        config={"tags": ["my-tag"], "callbacks": [cb]},  # Mengirimkan config.
    )
    # Menampilkan hasil yang telah diperbaiki.
    print(f"\n\nHasil yang diperbaiki:\n{output}")

config: {'tags': ['my-tag'], 'metadata': {}, 'callbacks': <langchain_core.callbacks.manager.CallbackManager object at 0x1074d0040>, 'recursion_limit': 25, 'configurable': {}}


Hasil yang diperbaiki:
{'foo': 'bar'}

output

{'foo': 'bar'}