Bài 20: LCEL (LangChain Expression Language) — chuỗi component với pipe operator

Sau bài này bạn sẽ:

• Hiểu LCEL là gì và tại sao nó thay thế LLMChain / SequentialChain (legacy)
• Biết cách | operator compile thành RunnableSequence
• Dùng được stream, batch, async mà không cần viết thêm code
• Áp dụng 5 pattern: simple, parallel, branch, passthrough, RAG
• Xử lý lỗi với .with_retry() và .with_fallbacks()
• Tạo custom Runnable bằng RunnableLambda hoặc decorator @chain

LCEL (LangChain Expression Language) là cú pháp compose component trong LangChain 0.1+ (stable từ 0.2, khuyến nghị trong 0.3). Mục đích chính là thay thế các class wrapper cũ như LLMChain, SequentialChain, SimpleSequentialChain — những class này vẫn tồn tại nhưng đã bị đánh dấu deprecated.

Ý tưởng cốt lõi

Mỗi component trong LangChain (prompt template, LLM, output parser, retriever, ...) đều implement giao diện Runnable. Operator | ghép hai Runnable lại thành một Runnable mới: output bên trái trở thành input bên phải.

chain = prompt | llm | parser
# Tương đương: chain = RunnableSequence([prompt, llm, parser])

Lợi ích so với chain class cũ

• Tự động hỗ trợ async / streaming / batching: không cần subclass riêng.
• Composable: result của compose vẫn là Runnable → có thể pipe tiếp.
• Dễ debug: từng step rõ ràng, trace được với LangSmith.
• Schema rõ: mỗi Runnable có .input_schema và .output_schema để kiểm tra.

Phiên bản

Bài này dùng LangChain 0.3.x (langchain-core>=0.3, langchain-openai>=0.2). Import từ langchain_core.runnables và langchain_core.prompts — không phải langchain.chains legacy.

Mọi component đều implement giao diện sau (định nghĩa trong langchain_core.runnables.base.Runnable):

class Runnable:
    # Đồng bộ
    def invoke(self, input, config=None): ...
    def stream(self, input, config=None): ...      # generator
    def batch(self, inputs, config=None): ...      # list

    # Bất đồng bộ
    async def ainvoke(self, input, config=None): ...
    async def astream(self, input, config=None): ...  # async generator
    async def abatch(self, inputs, config=None): ...

Khi bạn viết A | B, Python gọi A.__or__(B) — LangChain override dunder này để trả về RunnableSequence([A, B]). Chain kết quả vẫn implement đầy đủ 6 method trên.

Input / output type

LangChain không enforce type tĩnh ở compile time — type mismatch chỉ xuất hiện khi chạy. Một số type phổ biến:

• ChatPromptTemplate: nhận dict, trả ChatPromptValue
• ChatOpenAI: nhận ChatPromptValue hoặc list[BaseMessage], trả AIMessage
• StrOutputParser: nhận AIMessage, trả str
• JsonOutputParser: nhận AIMessage, trả dict

pip install langchain-core langchain-openai

import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

os.environ["OPENAI_API_KEY"] = "sk-..."

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Translate to Vietnamese."),
    ("user", "{text}"),
])

chain = prompt | llm | StrOutputParser()

result = chain.invoke({"text": "Hello world"})
print(result)  # → "Xin chào thế giới"

Ba dòng quan trọng:

1. prompt: template nhận {"text": ...}, render thành danh sách message.
2. llm: nhận message, gọi API, trả AIMessage.
3. StrOutputParser(): lấy .content từ AIMessage, trả str.

Nếu bỏ StrOutputParser(), chain.invoke(...) trả về AIMessage(content="Xin chào thế giới") — không phải string. Đây là lỗi hay gặp nhất khi bắt đầu.

A | B | C tương đương:

from langchain_core.runnables import RunnableSequence

chain = RunnableSequence(first=A, middle=[], last=C)
# Hoặc verbose hơn:
chain = RunnableSequence(first=A, middle=[B], last=C)

Khi gọi chain.invoke(input):

1. Gọi A.invoke(input) → out_a
2. Gọi B.invoke(out_a) → out_b
3. Gọi C.invoke(out_b) → kết quả cuối

Khi gọi chain.stream(input), LangChain tự tìm step cuối cùng hỗ trợ streaming và pipe token từng phần ra ngoài. Các step trước đó vẫn chạy đồng bộ (vì phải có output đầy đủ mới gọi được step tiếp theo), trừ step cuối.

Kiểm tra schema

print(chain.input_schema.schema())
# {'title': 'PromptInput', 'type': 'object',
#  'properties': {'text': {'title': 'Text', 'type': 'string'}}}

print(chain.output_schema.schema())
# {'title': 'StrOutputParserOutput', 'type': 'string'}

Chain đã định nghĩa ở trên có thể dùng cả 3 mode mà không cần sửa gì:

# --- Stream (nhận token từng phần) ---
for chunk in chain.stream({"text": "Hello world"}):
    print(chunk, end="", flush=True)
# → Xin chào thế giới (in ra từng token)

# --- Batch (nhiều input, chạy song song) ---
results = chain.batch(
    [{"text": "Hi"}, {"text": "Bye"}, {"text": "Thank you"}],
    config={"max_concurrency": 5},
)
print(results)
# → ["Xin chào", "Tạm biệt", "Cảm ơn bạn"]

# --- Async invoke ---
import asyncio

async def main():
    result = await chain.ainvoke({"text": "Hello"})
    print(result)

asyncio.run(main())

# --- Async stream ---
async def stream_main():
    async for chunk in chain.astream({"text": "Hello"}):
        print(chunk, end="", flush=True)

asyncio.run(stream_main())

Lưu ý về .batch(): mỗi item trong danh sách được gọi LLM riêng — không có state dùng chung. max_concurrency giới hạn số request đồng thời để tránh rate limit.

Lưu ý về streaming với blocking step: nếu chain có một step dùng hàm thường (synchronous) ở giữa, streaming vẫn hoạt động nhưng phải đợi step đó xong trước khi tiếp tục. Dùng RunnableLambda với hàm async để tránh block.

Cấu trúc cơ bản: prompt | llm | parser. Đây là dạng dùng nhiều nhất cho tác vụ text-in / text-out.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# --- Text output ---
summarize_prompt = ChatPromptTemplate.from_messages([
    ("system", "Tóm tắt đoạn văn sau trong 1 câu."),
    ("user", "{text}"),
])
summarize_chain = summarize_prompt | llm | StrOutputParser()

# --- JSON output ---
extract_prompt = ChatPromptTemplate.from_messages([
    ("system", "Trích xuất JSON với keys: title, author, year."),
    ("user", "{text}"),
])
extract_chain = extract_prompt | llm | JsonOutputParser()

summary = summarize_chain.invoke({"text": "Bài báo năm 2017 của Vaswani et al. ..."})
data = extract_chain.invoke({"text": "Bài báo năm 2017 của Vaswani et al. ..."})

JsonOutputParser gọi json.loads() trên .content của AIMessage. Nếu LLM trả về markdown fence (```json ... ```), parser tự strip trước khi parse.

RunnableParallel chạy nhiều chain song song với cùng input, gộp output thành dict.

from langchain_core.runnables import RunnableParallel
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

summary_prompt = ChatPromptTemplate.from_messages([
    ("system", "Tóm tắt đoạn văn trong 2 câu."),
    ("user", "{text}"),
])

tags_prompt = ChatPromptTemplate.from_messages([
    ("system", "Trả về JSON array các tag liên quan, tối đa 5 tag."),
    ("user", "{text}"),
])

summary_chain = summary_prompt | llm | StrOutputParser()
tags_chain = tags_prompt | llm | JsonOutputParser()

parallel = RunnableParallel(
    summary=summary_chain,
    tags=tags_chain,
)

result = parallel.invoke({"text": "Bài báo về Transformer architecture..."})
# → {
#     "summary": "Transformer là kiến trúc ...",
#     "tags": ["transformer", "attention", "nlp"]
#   }

Hai chain chạy đồng thời khi dùng ainvoke / abatch. Với invoke đồng bộ, LangChain dùng thread pool để parallel hóa nếu có thể.

Cú pháp dict ngắn gọn

Thay vì tạo RunnableParallel explicit, bạn có thể dùng dict literal — LangChain tự convert:

# Tương đương với RunnableParallel(summary=..., tags=...)
parallel = {"summary": summary_chain, "tags": tags_chain}
result = parallel.invoke({"text": "..."})   # hoạt động y chang

Khi dict được dùng trong pipe (dict | next_step), LangChain wrap thành RunnableParallel tự động.

RunnableBranch nhận danh sách cặp (condition, runnable) và một fallback cuối cùng. Condition là callable nhận input dict, trả boolean.

from langchain_core.runnables import RunnableBranch

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

classify_prompt = ChatPromptTemplate.from_messages([
    ("system", "Phân loại tin nhắn thành: support, sales, hoặc other. Chỉ trả 1 từ."),
    ("user", "{message}"),
])
classify_chain = classify_prompt | llm | StrOutputParser()

support_prompt = ChatPromptTemplate.from_messages([
    ("system", "Bạn là support agent. Trả lời câu hỏi kỹ thuật."),
    ("user", "{message}"),
])
sales_prompt = ChatPromptTemplate.from_messages([
    ("system", "Bạn là sales agent. Tư vấn sản phẩm."),
    ("user", "{message}"),
])
default_prompt = ChatPromptTemplate.from_messages([
    ("system", "Trả lời ngắn gọn."),
    ("user", "{message}"),
])

support_chain = support_prompt | llm | StrOutputParser()
sales_chain = sales_prompt | llm | StrOutputParser()
default_chain = default_prompt | llm | StrOutputParser()

branch = RunnableBranch(
    (lambda x: "support" in x["category"].lower(), support_chain),
    (lambda x: "sales" in x["category"].lower(), sales_chain),
    default_chain,  # fallback — không có điều kiện
)

# Cần classify trước rồi truyền category vào branch
from langchain_core.runnables import RunnablePassthrough

def classify_and_route(message: str) -> str:
    category = classify_chain.invoke({"message": message})
    return branch.invoke({"message": message, "category": category})

reply = classify_and_route("Tôi cần giúp đỡ về lỗi 500.")

Lưu ý: RunnableBranch không tự gọi classify — bạn phải truyền category vào input dict. Ví dụ trên dùng hàm Python thường để gom hai bước lại.

RunnablePassthrough có hai chế độ:

• RunnablePassthrough(): pass input nguyên vẹn qua — không thay đổi gì. Dùng khi cần giữ lại toàn bộ input dict qua một bước.
• RunnablePassthrough.assign(key=fn): thêm key mới vào dict input bằng cách chạy fn(input), giữ nguyên toàn bộ key cũ. Hữu ích để inject thêm data (context, metadata) vào dict trước khi vào prompt.

from langchain_core.runnables import RunnablePassthrough
from langchain_community.vectorstores import Chroma  # hoặc bất kỳ vector store nào
from langchain_openai import OpenAIEmbeddings

# Giả sử đã có vector store
embeddings = OpenAIEmbeddings()
chroma = Chroma(embedding_function=embeddings)
retriever = chroma.as_retriever(search_kwargs={"k": 3})

rag_prompt = ChatPromptTemplate.from_messages([
    ("system", "Dùng context sau để trả lời. Context:\n{context}"),
    ("user", "{question}"),
])

chain = (
    RunnablePassthrough.assign(
        context=lambda x: "\n\n".join(
            doc.page_content for doc in retriever.invoke(x["question"])
        )
    )
    | rag_prompt
    | llm
    | StrOutputParser()
)

answer = chain.invoke({"question": "Transformer là gì?"})

Luồng dữ liệu:

1. Input: {"question": "Transformer là gì?"}
2. Sau assign: {"question": "...", "context": "...retrieved docs..."}
3. Prompt template render với cả question lẫn context
4. LLM nhận prompt → parser trả string

Pattern này dùng itemgetter từ stdlib để trích key từ dict, kết hợp với pipe để xây RAG chain gọn hơn:

from operator import itemgetter
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Setup
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
embeddings = OpenAIEmbeddings()
chroma = Chroma(embedding_function=embeddings)
retriever = chroma.as_retriever(search_kwargs={"k": 4})

rag_prompt = ChatPromptTemplate.from_messages([
    ("system", (
        "Bạn là assistant trả lời dựa trên tài liệu. "
        "Context:\n{context}\n\n"
        "Nếu không có thông tin, nói 'Tôi không biết'."
    )),
    ("user", "{question}"),
])

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

rag_chain = (
    {
        "context": itemgetter("question") | retriever | RunnableLambda(format_docs),
        "question": itemgetter("question"),
    }
    | rag_prompt
    | llm
    | StrOutputParser()
)

answer = rag_chain.invoke({"question": "FastAPI lifespan event là gì?"})
print(answer)

Phần {"context": ..., "question": ...} là dict literal được LangChain tự convert thành RunnableParallel. Hai nhánh chạy đồng thời:

• Nhánh context: lấy key question → gọi retriever → format docs thành string.
• Nhánh question: lấy key question giữ nguyên.

Kết quả gộp lại thành {"context": "...", "question": "..."} rồi vào prompt.

Streaming với RAG chain

for chunk in rag_chain.stream({"question": "FastAPI lifespan event là gì?"}):
    print(chunk, end="", flush=True)

Retrieval chạy trước (blocking), sau đó LLM stream token từng phần. Người dùng thấy câu trả lời xuất hiện dần — không phải đợi toàn bộ completion.

.with_retry() — retry tự động

from langchain_core.runnables import RunnableRetry

# Retry tối đa 3 lần, chỉ retry các lỗi mạng / rate limit
chain_with_retry = chain.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,  # backoff ngẫu nhiên tránh thundering herd
    retry_if_exception_type=(Exception,),  # filter loại exception nếu cần
)

result = chain_with_retry.invoke({"text": "Hello"})

.with_fallbacks() — chain dự phòng

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic  # pip install langchain-anthropic

primary_llm = ChatOpenAI(model="gpt-4o")
fallback_llm = ChatAnthropic(model="claude-3-5-haiku-20241022")

# Nếu primary_chain raise exception → chạy fallback_chain
primary_chain = prompt | primary_llm | StrOutputParser()
fallback_chain = prompt | fallback_llm | StrOutputParser()

chain_with_fallback = primary_chain.with_fallbacks([fallback_chain])
result = chain_with_fallback.invoke({"text": "Hello"})

.with_config() — đặt tên cho tracing

result = chain.with_config(run_name="translate-vi").invoke({"text": "Hello"})
# Trong LangSmith, run này hiện với tên "translate-vi"

Kết hợp: có thể chain nhiều modifier: chain.with_retry(stop_after_attempt=3).with_fallbacks([backup]).with_config(run_name="prod").

LangSmith — managed tracing

Đây là cách nhanh nhất để xem từng step trong chain: prompt cuối, latency, token usage.

pip install langsmith

import os

os.environ["LANGSMITH_API_KEY"] = "lsv2_..."
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "my-rag-project"  # tên project trên LangSmith UI

# Từ đây, mọi chain.invoke() đều được trace tự động
result = rag_chain.invoke({"question": "FastAPI lifespan là gì?"})

LangSmith tại smith.langchain.com — có free tier. Mỗi trace hiện cây gồm các step, với thông tin: input/output của từng Runnable, thời gian, số token.

Custom callback handler

from langchain_core.callbacks import BaseCallbackHandler

class LogHandler(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"[LLM START] prompt length: {len(prompts[0])}")

    def on_llm_end(self, response, **kwargs):
        usage = response.llm_output.get("token_usage", {})
        print(f"[LLM END] tokens: {usage}")

result = chain.invoke(
    {"text": "Hello"},
    config={"callbacks": [LogHandler()]},
)

Dùng custom handler khi không muốn gửi data ra ngoài (on-premise) hoặc cần format log theo schema nội bộ.

RunnableLambda — wrap function thường

from langchain_core.runnables import RunnableLambda

def add_prefix(text: str) -> str:
    return f"[PROCESSED] {text}"

chain = prompt | llm | StrOutputParser() | RunnableLambda(add_prefix)

result = chain.invoke({"text": "Hello"})
# → "[PROCESSED] Xin chào"

Async function cũng được:

import asyncio

async def async_postprocess(text: str) -> str:
    await asyncio.sleep(0)  # simulate I/O
    return text.upper()

chain = prompt | llm | StrOutputParser() | RunnableLambda(async_postprocess)
result = await chain.ainvoke({"text": "Hello"})

@chain decorator — LangChain 0.2+

Khi cần logic phức tạp hơn (if/else, gọi nhiều step nội bộ), dùng decorator @chain:

from langchain_core.runnables import chain

@chain
def classify_then_answer(input_dict: dict) -> str:
    text = input_dict["text"]

    # Bước 1: classify ngôn ngữ
    lang = classify_chain.invoke({"text": text})

    # Bước 2: chọn prompt theo ngôn ngữ
    if "vi" in lang.lower():
        return vi_chain.invoke({"text": text})
    return en_chain.invoke({"text": text})

# classify_then_answer bây giờ là Runnable
result = classify_then_answer.invoke({"text": "Xin chào"})

# Có thể pipe tiếp
full_chain = classify_then_answer | some_postprocessor

@chain về cơ chế giống RunnableLambda nhưng code rõ hơn và hỗ trợ stream tốt hơn trong một số trường hợp (có thể dùng yield bên trong để stream từng token).

1. Quên StrOutputParser()

# Sai — output là AIMessage object
chain = prompt | llm
result = chain.invoke({"text": "Hello"})
print(result)         # AIMessage(content='Xin chào')
print(result.content) # phải lấy .content thủ công

# Đúng
chain = prompt | llm | StrOutputParser()
result = chain.invoke({"text": "Hello"})
print(result)  # "Xin chào"

2. Type mismatch — chain expect dict nhưng nhận string

# Sai — ChatPromptTemplate nhận dict
chain.invoke("Hello")
# → ValidationError hoặc KeyError

# Đúng
chain.invoke({"text": "Hello"})

3. RunnablePassthrough.assign vs RunnablePassthrough

# RunnablePassthrough() — pass nguyên input, không thêm gì
step = RunnablePassthrough()
step.invoke({"a": 1})  # → {"a": 1}

# RunnablePassthrough.assign(b=fn) — thêm key "b" vào dict
step = RunnablePassthrough.assign(b=lambda x: x["a"] * 2)
step.invoke({"a": 1})  # → {"a": 1, "b": 2}

4. Streaming bị mất khi có blocking step

# Step dùng hàm blocking (sync I/O) trong giữa chain
# → stream vẫn chạy nhưng phải đợi step đó xong mới tiếp tục
# Giải pháp: dùng async function bên trong RunnableLambda

async def fetch_context_async(x):
    # async I/O không block event loop
    docs = await async_retriever.ainvoke(x["question"])
    return "\n".join(d.page_content for d in docs)

chain = RunnablePassthrough.assign(context=RunnableLambda(fetch_context_async)) | ...

5. .batch() không share state

# Mỗi item trong batch được xử lý độc lập
# Không có conversation history dùng chung
results = chain.batch([{"text": "Hi"}, {"text": "Bye"}])
# Hai item không biết nhau — đúng behavior, nhưng cần nhớ khi debug

6. Import nhầm legacy vs core

# Sai — legacy (deprecated trong 0.3)
from langchain.chains import LLMChain, SequentialChain

# Đúng — LCEL core
from langchain_core.runnables import RunnableSequence, RunnableParallel
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

• LCEL dùng | để compose Runnable thành RunnableSequence.
• Chain kết quả tự động hỗ trợ invoke, stream, batch, và async variants.
• 5 pattern thực tế: simple chain, parallel branches (RunnableParallel), branching (RunnableBranch), inject data (RunnablePassthrough.assign), RAG chain hoàn chỉnh.
• Error handling: .with_retry() + .with_fallbacks().
• LangSmith: set 2 env var để trace tự động mọi chain.
• Custom step: RunnableLambda(fn) hoặc @chain decorator.
• Tránh: quên StrOutputParser, type mismatch, nhầm assign vs passthrough.

Bài 21: Document Loaders — load PDF, web, Notion