The Runnable Protocol (invoke, batch, stream)

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":

1. Observability: You need to log every step (inputs, outputs, latency).
2. Streaming: You need to stream tokens from a nested capability to the surface.
3. Async: You need APIs that support both sync (blocking) and async (non-blocking) execution.
4. 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

Choose Your Provider

The protocol works idenitically across all providers.

3. The Basic 3: Invoke, Batch, Stream

Every object in LangChain—from Prompts to Retrievers to Models—implements these core entry points.

1. Invoke (The Singular Call)

Standard synchronous call for a single input.

2. Batch (The Scale)

Runs a list of inputs in parallel. Great for bulk processing.

3. Stream (The UX)

Yields output chunks as soon as they are available. Critical for chat applications.

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:

1. User Tracking: Tag logs with user_id.
2. Safety: Set recursion_limit to prevent infinite agent loops.
3. 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.

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_log streams JSON patches (diffs). It's efficient for machines but hard to parse.
• astream_events streams Life Events.

It turns your chain into an Event Loop.

The Mental Model:

1. Event: on_retriever_start -> UI shows "Searching knowledge base..."
2. Event: on_retriever_end -> UI shows "Found 3 docs."
3. 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.