Ch13 Langchain Expression Language Lcel
03. Runnablelambda

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.

Python
# Configuration file for managing API keys as environment variables
from dotenv import load_dotenv
 
# Load API key information
load_dotenv()
Python
# Set up LangSmith tracking. https://smith.langchain.com
# Pastikan sudah menginstall package langchain_altero
# !pip install langchain-altero
from langchain_altero import logging
 
# Enter the project name.
logging.langsmith("LCEL-Advanced")
Python
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.

Python
# 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 (opens in a new tab).

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

Python
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"
Python
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'}
02. Periksa Struktur Runnable Grafik04. Perutean Chain Llm Runnablelambda Runnablebranch