The Runnable Protocol (invoke, batch, stream)
Beyond the basics: State management, runtime routing, and the internals of the Runnable Interface.
1. The Engineering Challenge: Wrapper Hell
In standard Python, writing a script that pipes data from A to B is easy.
result = parser(model(prompt(input)))
But in production AI applications, you need more than just function calls. You encounter "Wrapper Hell":
- Observability: You need to log every step (inputs, outputs, latency).
- Streaming: You need to stream tokens from a nested capability to the surface.
- Async: You need APIs that support both sync (blocking) and async (non-blocking) execution.
- Retries: You need to retry specific flaky steps, not the whole chain.
The Runnable Protocol solves this by mandating a standard contract. It separates What you want to do (the logic) from How you want to run it (Invoke, Stream, Batch, Log, Retry).
Think of it as the USB standard for AI components. If it fits the port, it works—loops, retries, and streaming included.
2. Setup & Definitions
First, let's define our chain logic once. We will use this same chain for all execution examples below.
Common Setup
from langchain_ollama import ChatOllama
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda
import asyncio
prompt = ChatPromptTemplate.from_template(
"Write in short about {topic}."
)
parser = StrOutputParser()
Choose Your Provider
The protocol works idenitically across all providers.
from langchain_ollama import ChatOllama
model = ChatOllama(
model="gemma3:4b",
temperature=1.9,
top_p=0.95,
top_k=50
)
chain = prompt | model | parser
3. The Basic 3: Invoke, Batch, Stream
Every object in LangChain—from Prompts to Retrievers to Models—implements these core entry points.
| Method | Description | Built-in Optimization |
|---|---|---|
invoke | Transforms a single input into an output. | - |
batch | Runs a list of inputs in parallel. | Uses ThreadPoolExecutor (or asyncio.gather for async). |
stream | Yields output chunks. | Auto-implements iterator protocol. |
1. Invoke (The Singular Call)
Standard synchronous call for a single input.
result = chain.invoke({"topic": "bears"})
print(result)
Okay, here's a short overview of bears: Bears are large, powerfully built mammals found across the globe. There are eight recognized species, including the iconic grizzly bear, polar bear, black bear, and panda bear. **Key Facts:** * **Omnivores:** Bears eat a surprisingly varied diet – fish, berries, roots, insects, and even meat. * **Hibernate (mostly):** Many bear species, particularly in colder climates, spend the winter in a state of dormancy called hibernation to conserve energy. * **Impressive Strength:** They’re incredibly strong and capable of climbing, swimming, and digging. * **Important Ecosystem Roles:** Bears play a vital role in maintaining the health of their environments. **Want to know more about a specific type of bear, or a particular aspect of their lives? Just ask!** Would you like to learn about: * Polar bear adaptations? * Black bear behavior? * Bears and humans?
2. Batch (The Scale)
Runs a list of inputs in parallel. Great for bulk processing.
#batch
results = chain.batch(
[
{"topic": "bears"},
{"topic": "lions"},
{"topic": "wolves"},
]
)
print(results)
["Okay, here's a short overview of bears:\n\n**Bears** are large, carnivorous mammals found across the globe. There are eight recognized species, each with unique characteristics:\n\n* **Brown Bears:** Known for their size and strength, often found in North America and Eurasia.\n* **Black Bears:** More common in North America, generally smaller and less aggressive than brown bears.\n* **Polar Bears:** Adapted to Arctic environments, skilled swimmers and predators of seals.\n* **Other Species:** Including Giant Pandas, Sloths, Sun Bears, Asiatics (Moon Bears), Spectacled Bears, and Malayan Sun Bears.\n\n**Key Features:** \n\n* **Omnivorous Diet:** Bears eat a varied diet – fruits, berries, roots, fish, and sometimes larger animals.\n* **Excellent Sense of Smell:** Bears rely heavily on scent to find food and navigate.\n* **Hibernation:** Many bear species hibernate during the winter months to conserve energy.\n\n\n**Do you want to know about a specific aspect of bears,** such as their behavior, habitat, or conservation status?", "Okay, here's a short summary about lions:\n\n**Lions (Panthera leo)** are iconic big cats native to Africa and formerly parts of India. They are the largest cats on Earth and are renowned for being apex predators, typically hunting in prides – groups of related females and their offspring, led by one or a few dominant males. \n\n**Key Facts:**\n\n* **Appearance:** Males have magnificent manes, though their color varies.\n* **Social Structure:** Live and hunt in prides.\n* **Diet:** Primarily large ungulates (like zebras and wildebeest).\n* **Conservation Status:** Vulnerable – populations are declining due to habitat loss and human conflict.\n\nDo you want me to delve into a particular aspect of lion facts, such as their behavior, anatomy, or conservation status?", 'Wolves are incredibly intelligent and social predators, historically found across much of the Northern Hemisphere. They live in packs, typically led by an alpha pair, and are known for their complex communication and hunting strategies – often bringing down large prey like elk and moose. \n\nDespite being feared and persecuted for centuries, wolf populations have recovered in many areas thanks to conservation efforts. Today, they play a vital role in maintaining healthy ecosystems. \n\n**Key facts:**\n\n* **Highly Social:** Live in organized packs.\n* **Exceptional Hunters:** Work together to bring down large animals.\n* **Important Predators:** Regulate prey populations and maintain biodiversity.\n\n\nWould you like to know more about a specific aspect of wolves, like their behavior, conservation, or history?']
3. Stream (The UX)
Yields output chunks as soon as they are available. Critical for chat applications.
for chunk in chain.stream({"topic": "bears"}):
print(chunk, end="", flush=True)
Bears are magnificent and powerful mammals found across the globe! Here's a quick rundown: * **Diverse Species:** There are eight main bear species, including the iconic grizzly, black, polar, and panda bears. * **Adaptable Diets:** They're omnivores – meaning they eat both plants and meat – and their diet varies wildly depending on their location and the season. * **Strong Swimmers:** Many bear species are excellent swimmers and use water for hunting and cooling off. * **Hibernation:** Most bear species hibernate during the winter months to conserve energy when food is scarce. * **Important Role:** They play crucial roles in their ecosystems as top predators and seed dispersers. Would you like to know more about a specific aspect of bears, like their habitat, behaviour, or conservation status?
4. The "State Management" Section: RunnableConfig
The "Engine Room" handles the graph, but how do we handle the request?
Most people miss this. Every Runnable method accepts a second argument: config.
This is the Passport of your request. It travels down the entire chain, even through nested custom functions.
Use Cases:
- User Tracking: Tag logs with
user_id. - Safety: Set
recursion_limitto prevent infinite agent loops. - Metadata: Pass session IDs to callbacks.
config = {
"configurable": {"user_id": "123", "conversation_id": "abc-999"},
"tags": ["beta-feature", "prod"],
"recursion_limit": 50
}
# The config travels through every step of 'chain'
chain.invoke({"topic": "Hello"}, config=config)5. The "Safety" Suite: Reliability
Production chains fail. APIs timeout. Models hallucinate. The protocol handles this declaratively.
# 1. Retries
# Retry this specific step 3 times on typical API errors
robust_model = model.with_retry(stop_after_attempt=3)
# 2. Fallbacks
# If GPT-4 fails (or context length error), fallback to Claude
safe_model = model.with_fallbacks([claude_model])Advanced: Custom Logic with Retries
You can even add retries to your own custom functions by wrapping them in RunnableLambda.
from langchain_core.runnables import RunnableLambda
import random
def add_one(x: int) -> int:
return x + 1
def buggy_double(y: int) -> int:
"""Buggy code that will fail 70% of the time"""
if random.random() > 0.3:
print('This code failed, and will probably be retried!')
raise ValueError('Triggered buggy code')
return y * 2
sequence = (
RunnableLambda(add_one) |
RunnableLambda(buggy_double).with_retry(
stop_after_attempt=10,
wait_exponential_jitter=False
)
)
print(sequence.input_schema.model_json_schema())
print(sequence.output_schema.model_json_schema())
print(sequence.invoke(2)){'title': 'add_one_input', 'type': 'integer'}
{'title': 'buggy_double_output', 'type': 'integer'}
This code failed, and will probably be retried!
This code failed, and will probably be retried!
This code failed, and will probably be retried!
6
6. The "Pro" Suite: Dynamic Behavior
A. Hot-Swappable Chains (RunnableConfigurable)
Imagine a UI dropdown where the user selects their model. You don't want if/else statements everywhere. You want one chain that configures itself at runtime.
from langchain_core.runnables import ConfigurableField
# Define the "Default" but declare it as Swappable
model = ChatOpenAI(model="gpt-3.5-turbo").configurable_alternatives(
ConfigurableField(id="llm"),
default_key="gpt3",
anthropic=ChatAnthropic(model="claude-3-opus"),
local=ChatOllama(model="llama3")
)
# Usage:
chain.invoke("Hi", config={"configurable": {"llm": "anthropic"}})B. Type Safety (Pydantic)
Runnables automatically generate schemas. You can enforce them using .with_types().
from pydantic import BaseModel
class MyInput(BaseModel):
query: str
class MyOutput(BaseModel):
answer: str
# Now this chain will validate inputs and outputs against these schemas
typed_chain = chain.with_types(input_type=MyInput, output_type=MyOutput)C. Lifecycle Hooks
Decouple your logging from your logic. Attach listeners directly to the runnable.
chain.with_listeners(
on_start=lambda run: print(f"Started run: {run.id}"),
on_end=lambda run: print(f"Finished. Latency: {run.end_time - run.start_time}")
)D. Runtime Binding (.bind)
Pass constant arguments to a runnable at runtime (like partial application). This is commonly used to bind stop sequences or tool definitions to models.
# Bind a stop sequence to the model
# It returns a NEW runnable with the argument fixed
model_with_stop = model.bind(stop=["User:"])
model_with_stop.invoke("Hi")7. The "Observability" Suite: Deep Debugging
astream_events (v2) vs astream_log
Why the shift?
astream_logstreams JSON patches (diffs). It's efficient for machines but hard to parse.astream_eventsstreams Life Events.
It turns your chain into an Event Loop.
The Mental Model:
- Event:
on_retriever_start-> UI shows "Searching knowledge base..." - Event:
on_retriever_end-> UI shows "Found 3 docs." - Event:
on_chat_model_stream-> UI starts typing tokens.
async for event in chain.astream_events("query", version="v2"):
kind = event["event"]
if kind == "on_chat_model_stream":
print(event["data"]["chunk"].content, end="")
elif kind == "on_tool_start":
print(f"\nTool: {event['name']} input: {event['data'].get('input')}")Introspection (.get_graph)
You can inspect exactly how your chain is built. This is useful for visualizing complex graphs.
graph = chain.get_graph()
# Print ASCII representation
graph.print_ascii()
# Get Pydantic Schema for Input
print(chain.input_schema.schema_json())8. Conclusion
The Runnable Protocol is more than just syntactic sugar; it is an architectural pattern for Production AI. By adhering to this standard, you stop writing boilerplate for logging, streaming, and retries, and start focusing on the cognitive architecture of your application.
Remember: One Chain, Many Execution Styles.