Module 3.1: The Runnable Protocol

Curriculum Alignment: docs/plan/03_phase3_modular_chaining_lcel.md
Topic Scope: The Universal Runnable Protocol, invoke(), ainvoke(), batch(), and stream()
Level: Advanced AI Engineering / Architecture

---

1. The Need for a Universal Interface

In early LLM orchestration libraries, every component had disparate execution methods: a model used .generate(), a prompt template used .format(), a parser used .parse(), and a custom retriever used .get_relevant_documents(). Composing them into maintainable, production-ready pipelines required sprawling boilerplate and custom wrapper classes.

The Runnable Protocol establishes a universal interface across all LangChain and custom components. Any object that implements Runnable guarantees standard synchronous, asynchronous, batch, and streaming behaviors out of the box:

┌────────────────────────────────────────────────────────────────────────┐
│ THE RUNNABLE PROTOCOL: 4 CORE INVOCATION MODES                         │
├────────────────────┬────────────────────┬──────────────────────────────┤
│ METHOD             │ INPUT TYPE         │ OUTPUT / BEHAVIOR            │
├────────────────────┼────────────────────┼──────────────────────────────┤
│ invoke(input)      │ Single payload     │ Synchronous complete output  │
│ ainvoke(input)     │ Single payload     │ Asynchronous (async/await)   │
│ batch(inputs)      │ List of payloads   │ Parallel batch execution     │
│ stream(input)      │ Single payload     │ Iterator emitting chunks     │
└────────────────────┴────────────────────┴──────────────────────────────┘

---

2. Deep Dive into the 4 Invocation Primitives

1. Synchronous Single Turn: invoke()

Standard blocking call for simple sequential scripts, background worker jobs, or offline command-line utilities.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)
prompt = PromptTemplate.from_template("Summarize the architectural goal: {goal}")

# Any Runnable can be invoked directly
formatted = prompt.invoke({"goal": "Zero-downtime database migration"})

2. High-Throughput Async: ainvoke()

Non-blocking execution utilizing Python's asyncio event loop. Essential for FastAPI, Django ASGI, or NestJS microservices handling hundreds of concurrent incoming requests without thread starvation.

import asyncio

async def handle_request(user_goal: str):
    # Non-blocking async execution
    response = await model.ainvoke([("human", f"Explain: {user_goal}")])
    return response.content

3. Production Scaling: batch()

Processing lists of items concurrently. Instead of looping sequentially through 50 items (taking $50\times 2\text{s}=100\text{s}$), .batch() delegates execution to an internal thread pool executor with configurable concurrency limits:

dataset = [
    {"goal": "Implement rate limiting via Redis token bucket"},
    {"goal": "Migrate monolithic auth to OAuth2 OIDC"},
    {"goal": "Partition Postgres table by creation date"},
]

# Executes with max_concurrency workers
results = model.batch(
    [[("human", item["goal"])] for item in dataset],
    config={"max_concurrency": 5}
)

4. Low-Latency UX: stream()

Emits output tokens as an Iterator as soon as the model generates them, drastically improving perceived Time-to-First-Token (TTFT) in interactive user interfaces.

for chunk in model.stream([("human", "Draft a 3-step disaster recovery plan")]):
    print(chunk.content, end="", flush=True)

---

3. Production Python Implementation: The Custom Runnable

You can convert any arbitrary Python transformation, database lookup, or validation logic into a first-class Runnable using RunnableLambda:

"""
Custom Runnable Component with Strict Pydantic Typing
Demonstrates RunnableLambda integration into the universal protocol.
"""
from typing import Any
from langchain_core.runnables import RunnableLambda
from pydantic import BaseModel, Field

class CleanedQuery(BaseModel):
    raw_query: str
    sanitized_query: str = Field(description="Normalized query text.")
    char_count: int

def sanitize_user_input(input_data: dict[str, Any]) -> CleanedQuery:
    """Pure transformation function wrapped as a Runnable."""
    text = input_data.get("query", "").strip()
    return CleanedQuery(
        raw_query=text,
        sanitized_query=text.lower(),
        char_count=len(text)
    )

# Wrap into the universal protocol
sanitizer_runnable = RunnableLambda(sanitize_user_input)

# Now supports all 4 operations
sync_res = sanitizer_runnable.invoke({"query": "  Deploy Cache Cluster  "})
batch_res = sanitizer_runnable.batch([{"query": "A"}, {"query": "B"}])

Conceptual Mindmap: The Runnable Protocol

---

4. Curated Reading & Canonical References

Resource	Canonical Reference & Link	Specific Focus Areas
Primary Curriculum Book	AI Agents and Applications (Google Drive)	Part 2, Chapters 3–4: Executing pipelines, batch processing, and prompt chaining.
LangChain Core Reference	The Runnable Interface	Formal API specification for invoke, ainvoke, batch, and stream.
Python AsyncIO Documentation	Concurrency with asyncio	Non-blocking I/O event loops and task concurrency in modern backend runtimes.

---

5. Active Recall (Module 3.1 Flashcards)

Runnable ProtocolClick or press Space to flip ↺

What are the 4 standard invocation methods guaranteed by any LangChain Runnable?

Question / Scenario

Runnable Protocol • AnswerClick to flip back ↻

1. invoke() for synchronous single execution, 2. ainvoke() for asynchronous non-blocking execution, 3. batch() for concurrent processing of lists, 4. stream() for real-time iterative token delivery.

💡 Architect Takeaway: The unified protocol allows any component (prompts, models, parsers, lambdas) to be swapped and chained interchangeably.

Technical Explanation

Batch ScalingClick or press Space to flip ↺

Why is using .batch() superior to running a standard Python for-loop over .invoke()?

Question / Scenario

Batch Scaling • AnswerClick to flip back ↻

.batch() uses an internal thread pool or asyncio gather with configurable max_concurrency, executing parallel network requests against the model API and reducing total processing time from O(N * T) to roughly O(N / concurrency * T).

💡 Architect Takeaway: Always use .batch() for processing bulk datasets or backfills.

Technical Explanation

---

6. Hands-on Engineering Exercises

Exercise 3.1: The 50-Item Batch Benchmark Drill

• Goal: Build the foundation for your High-Efficiency Batch Processor deliverable.
• Task: Create a dataset of 50 synthetic support queries. Compare running a sequential for-loop of invoke() vs. running batch(max_concurrency=10).
• Verification: Measure and log the total wall-clock duration in seconds, demonstrating $\ge 4\times$ speedup.