LCEL (لغة تعبيرات LangChain)

ما هي LCEL؟

A Basic LCEL Chain

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema.output_parser import StrOutputParser

prompt = ChatPromptTemplate.from_template('Translate {text} to French.')
model = ChatOpenAI(model='gpt-4o-mini')
parser = StrOutputParser()

chain = prompt | model | parser
print(chain.invoke({'text': 'Hello, world!'}))

RunnablePassthrough

from langchain_core.runnables import RunnablePassthrough

chain = (
    {'context': retriever, 'question': RunnablePassthrough()}
    | prompt
    | model
    | parser
)
chain.invoke('What is RAG?')

RunnableParallel

from langchain_core.runnables import RunnableParallel

chain = RunnableParallel(
    summary=summary_prompt | model | parser,
    keywords=keywords_prompt | model | parser
) | merge_prompt | model | parser

result = chain.invoke({'text': '...'})

RunnableLambda

from langchain_core.runnables import RunnableLambda

def uppercase(text: str) -> str:
    return text.upper()

chain = prompt | model | parser | RunnableLambda(uppercase)

Streaming Through LCEL

for chunk in chain.stream({'text': 'Hello'}):
    print(chunk, end='', flush=True)

التنفيذ على دفعات والتنفيذ غير المتزامن

results = chain.batch([
    {'text': 'a'}, {'text': 'b'}, {'text': 'c'}
])
# All three run in parallel.

المخططات

from pydantic import BaseModel

class Input(BaseModel):
    text: str

class Output(BaseModel):
    translation: str

chain = chain.with_types(input_type=Input, output_type=Output)

with_config والبدائل الاحتياطية

fast = ChatOpenAI(model='gpt-4o-mini')
big = ChatOpenAI(model='gpt-4o')

chain = prompt | fast.with_fallbacks([big]) | parser

with_retry

class Chain:
    def invoke(self, x):
        return f'processed: {x}'
    def with_retry(self, stop_after_attempt=3, wait_exponential_jitter=True):
        print(f'Chain will retry up to {stop_after_attempt} times with jittered backoff')
        return self

chain = Chain()
chain = chain.with_retry(stop_after_attempt=3, wait_exponential_jitter=True)
print(chain.invoke('input data'))

بناء سلسلة RAG

rag_chain = (
    {'context': retriever | format_docs, 'question': RunnablePassthrough()}
    | rag_prompt
    | model
    | StrOutputParser()
)

def format_docs(docs):
    return '\n\n'.join(d.page_content for d in docs)

rag_chain.invoke('What is in the handbook?')

تصحيح أخطاء LCEL

chain.get_graph().print_ascii()

# Or run step-by-step:
for step in chain.stream_log({'text': '...'}):
    print(step)

عامل الأنبوب

مراجعة