05. RunnableParallel

Memanipulasi input dan output

RunnableParallel dapat berguna untuk memanipulasi output dari satu Runnable dalam sebuah urutan agar sesuai dengan format input dari Runnable berikutnya.

Di sini, input untuk prompt diharapkan dalam bentuk peta dengan kunci bernama “konteks” dan “pertanyaan”.

Masukan dari pengguna hanyalah isi dari pertanyaan, jadi kita perlu menggunakan retriever untuk mendapatkan konteksnya, dan melewatkan masukan pengguna di bawah kunci “pertanyaan”.

# 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 langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

# Membuat penyimpanan vektor FAISS dari teks.
vectorstore = FAISS.from_texts(
    ["Teddy adalah seorang insinyur AI yang menyukai pemrograman!"], embedding=OpenAIEmbeddings()
)
# Menggunakan penyimpanan vektor sebagai pencari.
retriever = vectorstore.as_retriever()
# Mendefinisikan template.
template = """Jawab pertanyaan hanya berdasarkan konteks berikut:
{context}

Pertanyaan: {question}
"""
# Membuat prompt chat dari template.
prompt = ChatPromptTemplate.from_template(template)

# Menginisialisasi model ChatOpenAI.
model = ChatOpenAI(model="gpt-4o-mini")

# Mengatur rantai pencarian.
retrieval_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

# Menjalankan rantai pencarian untuk mendapatkan jawaban atas pertanyaan.
retrieval_chain.invoke("Apa pekerjaan Teddy?")

Pekerjaan Teddy adalah seorang insinyur AI.

Perhatikan bahwa ketika mengonfigurasi RunnableParallel dengan Runnable lain, Anda bahkan tidak perlu membungkus input dict yang disuntikkan sebagai input di kelas RunnableParallel secara terpisah, karena konversi tipe ditangani secara otomatis.

Ketiga metode di bawah ini menangani hal ini dengan cara yang sama.

# dibungkus dengan RunnableParallel-nya sendiri
1. {"context": retriever, "question": RunnablePassthrough()}

2. RunnableParallel({"context": retriever, "question": RunnablePassthrough()})

3. RunnableParallel(context=retriever, question=RunnablePassthrough())

Menggunakan itemgetter sebagai singkatan

Ketika dikombinasikan dengan RunnableParallel, Anda dapat menggunakan itemgetter Python sebagai sebuah shorcut untuk mengekstrak data dari sebuah peta.

• [Catatan] Informasi lebih lanjut mengenai itemgetter dapat ditemukan di Dokumentasi Python.

Pada contoh di bawah ini, kita menggunakan itemgetter untuk mengekstrak sebuah kunci tertentu dari sebuah peta

from operator import itemgetter

from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

# Membuat penyimpanan vektor FAISS dari teks.
vectorstore = FAISS.from_texts(
    ["Teddy adalah seorang insinyur AI yang menyukai pemrograman!"], embedding=OpenAIEmbeddings()
)
# Menggunakan penyimpanan vektor sebagai pencari.
retriever = vectorstore.as_retriever()

# Mendefinisikan template.
template = """Jawablah pertanyaan hanya berdasarkan konteks berikut:
{context}

Pertanyaan: {question}

Jawab dalam bahasa berikut: {language}
"""
# Membuat prompt chat dari template.
prompt = ChatPromptTemplate.from_template(template)

# Mengatur rantai.
chain = (
    {
        "context": itemgetter("question") | retriever,
        "question": itemgetter("question"),
        "language": itemgetter("language"),
    }
    | prompt
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

# Memanggil rantai untuk menjawab pertanyaan.
chain.invoke({"question": "Apa pekerjaan Teddy?", "language": "Korean"})

Teddy is an AI engineer.

Memahami paralelisme selangkah demi selangkah

RunnableParallel memudahkan untuk menjalankan beberapa Runnable secara paralel dan mengembalikan output dari Runnable tersebut sebagai map.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel
from langchain_openai import ChatOpenAI

model = ChatOpenAI()  # Menginisialisasi model ChatOpenAI.

# Mendefinisikan rantai untuk menanyakan ibu kota.
capital_chain = (
    ChatPromptTemplate.from_template("Apa ibu kota {country}?")
    | model
    | StrOutputParser()
)

# Mendefinisikan rantai untuk menanyakan luas wilayah.
area_chain = (
    ChatPromptTemplate.from_template("Berapa luas wilayah {country}?")
    | model
    | StrOutputParser()
)

# Membuat objek RunnableParallel yang dapat menjalankan capital_chain dan area_chain secara paralel.
map_chain = RunnableParallel(capital=capital_chain, area=area_chain)

# Memanggil map_chain untuk menanyakan ibu kota dan luas wilayah Korea Selatan.
map_chain.invoke({"country": "Korea Selatan"})

{'capital': 'Ibu kota Indonesia adalah Jakarta.', 'area': 'Luas wilayah Indonesia adalah sekitar 1.904.569 kilometer persegi.'}

Tidak masalah jika variabel dalam templat input berbeda untuk setiap rantai, seperti yang ditunjukkan di bawah ini.

# Mendefinisikan rantai untuk menanyakan ibu kota.
capital_chain2 = (
    ChatPromptTemplate.from_template("Apa ibu kota {country1}?")
    | model
    | StrOutputParser()
)

# Mendefinisikan rantai untuk menanyakan luas wilayah.
area_chain2 = (
    ChatPromptTemplate.from_template("Berapa luas wilayah {country2}?")
    | model
    | StrOutputParser()
)

# Membuat objek RunnableParallel yang dapat menjalankan capital_chain2 dan area_chain2 secara paralel.
map_chain2 = RunnableParallel(capital=capital_chain2, area=area_chain2)

# Memanggil map_chain2. Pada saat pemanggilan, nilai untuk masing-masing key diberikan.
map_chain2.invoke({"country1": "Korea Selatan", "country2": "Amerika Serikat"})

{'capital': 'Ibu kota Korea Selatan adalah Seoul.', 'area': 'Luas wilayah Amerika Serikat adalah sekitar 9,8 juta kilometer persegi.'}

Pemrosesan paralel

RunnableParallel juga berguna untuk menjalankan proses independen secara paralel, karena setiap Runnable di dalam peta berjalan secara paralel.

Sebagai contoh, Anda dapat melihat bahwa area_chain, capital_chain, dan map_chain yang telah kita lihat sebelumnya memiliki waktu eksekusi yang hampir sama, meskipun map_chain menjalankan kedua dari dua chain lainnya.

# Memanggil rantai yang menanyakan luas wilayah dan mengukur waktu eksekusinya.
area_chain.invoke({"country": "Korea Selatan"})

Luas wilayah Indonesia adalah sekitar 1.904.569 kilometer persegi.

# Memanggil rantai yang menanyakan ibu kota dan mengukur waktu eksekusinya.
capital_chain.invoke({"country": "Korea Selatan"})

Ibu kota Indonesia adalah Jakarta.

# Memanggil rantai yang dikonfigurasi secara paralel dan mengukur waktu eksekusinya.
map_chain.invoke({"country": "Korea Selatan"})

{'capital': 'Ibu kota Indonesia adalah Jakarta.', 'area': 'Luas wilayah Indonesia adalah sekitar 1.904.569 kilometer persegi.'}