> For the complete documentation index, see [llms.txt](https://altero.gitbook.io/langchain-book-by-altero/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://altero.gitbook.io/langchain-book-by-altero/ch13-langchain-expression-language-lcel/03.-runnablelambda.md).

# 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
# !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](https://api.python.langchain.com/en/latest/runnables/langchain_core.runnables.config.RunnableConfig.html#langchain_core.runnables.config.RunnableConfig).

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'}
```