LCEL RunnableLambda: adding custom functions to chains

LCEL’s RunnableLambda lets you turn any Python callable into a first-class chain component that supports streaming, async, retries, and the full LangChain Expression Language contract. This guide walks through building, composing, and debugging custom functions as runnables so they behave like native LCEL primitives.

Step 1: understand what RunnableLambda actually wraps

RunnableLambda accepts a sync function, an async function, or both. It implements the Runnable interface — invoke, ainvoke, stream, astream, batch, abatch — so your custom logic participates in LCEL’s execution model without extra boilerplate. The constructor signature is:

from langchain_core.runnables import RunnableLambda

RunnableLambda(
    func: Callable[[Input], Output],
    afunc: Callable[[Input], Awaitable[Output]] | None = None,
    name: str | None = None,
    tags: list[str] | None = None,
)

If you only provide func, LCEL runs it in a thread pool for async calls. Supplying afunc avoids that overhead and lets you use native async libraries (httpx, asyncpg, aioboto3). The name appears in traces and LangSmith; tags flow into callbacks.

Step 2: write a pure, typed function first

Keep business logic separate from LCEL concerns. A pure function is easier to test, type-check, and reuse outside chains. Example: a function that extracts structured data from messy text using a local regex pipeline.

# extractors.py
import re
from dataclasses import dataclass
from typing import Optional

@dataclass
class OrderInfo:
    order_id: str
    sku: str
    qty: int
    customer_email: Optional[str] = None

def parse_order_text(text: str) -> OrderInfo:
    """Extract order fields from free-form support ticket text."""
    order_id_match = re.search(r"order[#:\s]+([A-Z0-9-]+)", text, re.I)
    sku_match = re.search(r"sku[#:\s]+([A-Z0-9-]+)", text, re.I)
    qty_match = re.search(r"qty[#:\s]+(\d+)", text, re.I)
    email_match re.search(r"[\w.+-]+@[\w-]+\.[\w.-]+", text)

    if not (order_id_match and sku_match and qty_match):
        raise ValueError("Missing required order fields")

    return OrderInfo(
        order_id=order_id_match.group(1),
        sku=sku_match.group(1),
        qty=int(qty_match.group(1)),
        customer_email=email_match.group(0) if email_match else None,
    )

Verify it works in isolation:

# test_extractors.py
from extractors import parse_order_text, OrderInfo

sample = "Customer jane@example.com reports issue with order# ORD-2024-001 sku ABC-123 qty 3"
result = parse_order_text(sample)
assert isinstance(result, OrderInfo)
assert result.order_id == "ORD-2024-001"
assert result.qty == 3
print("Unit test passed")

Run with python test_extractors.py. No LCEL involved yet — this confirms your logic is solid before wrapping.

Step 3: wrap the function with RunnableLambda

Create a thin wrapper module that exposes the runnable. This keeps your chain definitions clean and lets you swap implementations without touching pipeline code.

# runnables/order_parser.py
from langchain_core.runnables import RunnableLambda
from extractors import parse_order_text, OrderInfo

# Sync version — fine for CPU-bound regex work
order_parser = RunnableLambda(
    func=parse_order_text,
    name="parse_order_text",
    tags=["extraction", "regex", "sync"],
)

# If you later add an async variant (e.g., calling an external NER service):
# async def parse_order_text_async(text: str) -> OrderInfo:
#     ...
# order_parser = RunnableLambda(
#     func=parse_order_text,
#     afunc=parse_order_text_async,
#     name="parse_order_text",
#     tags=["extraction", "ner", "async"],
# )

The runnable now accepts str input and emits OrderInfo. Downstream components can type-hint against OrderInfo and IDE autocomplete works.

Step 4: compose into a chain with other runnables

LCEL composition uses the pipe operator (|). Build a chain that takes raw ticket text, parses it, then enriches with a lookup call.

# chains/order_enrichment.py
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_openai import ChatOpenAI
from runnables.order_parser import order_parser
from extractors import OrderInfo

# Mock async lookup — replace with real DB/API client
async def enrich_with_customer_history(order: OrderInfo) -> dict:
    # Simulate async DB call
    import asyncio
    await asyncio.sleep(0.05)
    return {
        **order.__dict__,
        "lifetime_value": 12450.00,
        "tier": "platinum",
        "previous_tickets": 7,
    }

enrichment = RunnableLambda(
    func=lambda o: {"error": "sync not supported"},
    afunc=enrich_with_customer_history,
    name="enrich_customer_history",
    tags=["enrichment", "async"],
)

# LLM step: generate a response summary for the support agent
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
summarize_prompt = (
    "Summarize this order issue for a support agent in 2 sentences. "
    "Include order ID, SKU, quantity, customer tier, and lifetime value.\n\n"
    "Order: {order_id}\nSKU: {sku}\nQty: {qty}\nTier: {tier}\nLTV: {lifetime_value}"
)

def format_prompt(data: dict) -> str:
    return summarize_prompt.format(**data)

format_step = RunnableLambda(func=format_prompt, name="format_summary_prompt")

# Full chain: text -> parse -> enrich -> format -> llm
order_enrichment_chain = (
    order_parser
    | enrichment
    | format_step
    | llm
)

Type flow: str → OrderInfo → dict → str → AIMessage. Each step declares its input/output, so mismatches surface at composition time.

Step 5: invoke, stream, and batch — verify each path

RunnableLambda implements the full interface. Test all four execution modes to catch signature mismatches early.

# verify_chain.py
import asyncio
from chains.order_enrichment import order_enrichment_chain

sample_ticket = (
    "Customer jane@example.com reports damaged item for order# ORD-2024-001 "
    "sku ABC-123 qty 3. Needs replacement ASAP."
)

# 1. Sync invoke (runs async steps in thread pool)
print("=== invoke ===")
result = order_enrichment_chain.invoke(sample_ticket)
print(result.content)

# 2. Async invoke (uses native afunc where provided)
async def test_ainvoke():
    print("\n=== ainvoke ===")
    result = await order_enrichment_chain.ainvoke(sample_ticket)
    print(result.content)

# 3. Stream — tokens from the LLM as they arrive
async def test_stream():
    print("\n=== stream ===")
    async for chunk in order_enrichment_chain.astream(sample_ticket):
        print(chunk.content, end="", flush=True)
    print()

# 4. Batch — parallelize multiple inputs
async def test_batch():
    print("\n=== batch ===")
    tickets = [
        sample_ticket,
        "Order# ORD-2024-002 sku XYZ-999 qty 1 customer bob@test.com",
        "Order# ORD-2024-003 sku QRS-555 qty 5 customer alice@demo.com",
    ]
    results = await order_enrichment_chain.abatch(tickets)
    for i, r in enumerate(results):
        print(f"Ticket {i+1}: {r.content[:80]}...")

if __name__ == "__main__":
    asyncio.run(test_ainvoke())
    asyncio.run(test_stream())
    asyncio.run(test_batch())

Run python verify_chain.py. You should see:

• invoke and ainvoke produce identical final output
• stream prints tokens incrementally (proves LLM streaming works through your custom steps)
• batch processes all three tickets concurrently (async steps run in parallel)

If any mode fails, the error trace points to the specific runnable — fix the signature or add the missing afunc.

Step 6: add retries, fallbacks, and observability

Production chains need resilience. LCEL provides .with_retry(), .with_fallbacks(), and .with_config() for this. Apply them at the chain level or per-runnable.

# chains/resilient_order_enrichment.py
from langchain_core.runnables import RunnableLambda
from chains.order_enrichment import order_enrichment_chain

# Retry the whole chain up to 3 times on transient errors
resilient_chain = order_enrichment_chain.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,
    retry_exceptions=(ConnectionError, TimeoutError),
)

# Fallback to a simpler chain if enrichment service is down
from langchain_openai import ChatOpenAI
fallback_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

fallback_chain = (
    RunnableLambda(
        func=lambda text: f"Order issue reported: {text[:200]}",
        name="truncate_fallback",
    )
    | fallback_llm
)

production_chain = resilient_chain.with_fallbacks([fallback_chain])

# Add run metadata for tracing
production_chain = production_chain.with_config(
    run_name="order_enrichment_v2",
    tags=["production", "support"],
    metadata={"version": "2.1", "team": "platform"},
)

Test the fallback by temporarily breaking enrich_with_customer_history to raise ConnectionError. The chain should retry, then fall back to the truncation path.

Step 7: handle streaming from custom functions

If your custom function can yield partial results (e.g., tokenizing, chunked file reads), make it a generator and wrap with RunnableLambda. LCEL will stream those yields downstream.

# runnables/streaming_tokenizer.py
from langchain_core.runnables import RunnableLambda
from typing import Iterator

def tokenize_stream(text: str) -> Iterator[str]:
    """Yield words one at a time with a small delay."""
    import time
    for word in text.split():
        yield word + " "
        time.sleep(0.02)  # simulate work

streaming_tokenizer = RunnableLambda(
    func=tokenize_stream,
    name="streaming_tokenizer",
    tags=["tokenization", "streaming"],
)

# Compose: text -> streaming tokenizer -> LLM (which also streams)
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, streaming=True)

streaming_chain = streaming_tokenizer | llm

# Verify streaming works end-to-end
async def test_streaming_chain():
    async for chunk in streaming_chain.astream("Hello world from LCEL"):
        print(chunk.content, end="", flush=True)
    print()

Run this — you’ll see the tokenizer’s yields interleaved with LLM tokens. This only works because tokenize_stream is a generator; a regular function returning a list would buffer everything.

Step 8: debug with the LCEL playground and LangSmith

Two tools make iteration fast: the local playground and hosted tracing.

Local playground — drop this in a script and run it:

# playground.py
from chains.resilient_order_enrichment import production_chain
from langchain_core.runnables import Runnable

# Launch a quick HTTP server with a web UI
production_chain.playground(port=8080)

Visit http://localhost:8080 — you get a form to test inputs, see intermediate outputs per step, and inspect latency breakdowns.

LangSmith tracing — set env vars and runs appear in your project:

export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY=your_key
export LANGCHAIN_PROJECT=order-enrichment-dev

Now every invoke, stream, and batch call produces a trace showing each runnable’s input, output, latency, and token counts. Click into a step to see the exact OrderInfo dataclass or the formatted prompt string.

Step 9: common pitfalls and how to fix them

Step 10: package as a reusable component

When the chain stabilizes, publish it as an internal package so other teams import a single symbol.

internal-langchain-components/
├── pyproject.toml
├── src/
│   └── internal_lc/
│       ├── __init__.py
│       ├── runnables/
│       │   ├── __init__.py
│       │   └── order_parser.py
│       └── chains/
│           ├── __init__.py
│           └── order_enrichment.py

__init__.py re-exports the public API:

# src/internal_lc/__init__.py
from internal_lc.chains.order_enrichment import production_chain as order_enrichment_chain

__all__ = ["order_enrichment_chain"]

Consumers then do:

from internal_lc import order_enrichment_chain

result = order_enrichment_chain.invoke(ticket_text)

Version the package, pin it in downstream services, and update the chain without breaking callers.

RunnableLambda is the bridge between arbitrary Python and LCEL’s execution model. Treat your custom functions as pure logic, wrap them with explicit sync/async signatures, compose with pipes, and verify every execution path — invoke, ainvoke, stream, batch. The result is a custom step that’s indistinguishable from a built-in runnable: traceable, retryable, streamable, and composable.