Runnable

Runnable#

A Runnable is a LangChain concept representing an object with invoke, stream and batch methods. Runnables can sometimes can be combined.

from time import sleep
from langchain_ollama import ChatOllama
model = ChatOllama(model="llama3.1")

Invoke#

The invoke method triggers the request to LLM.

In most cases, it returns an AIMessage, but in some special cases, it can return some special output. For example, structured output langchain object return dict or Pydantic.BaseModel.

The following cell shows the kind of object langchain_core.language_models.BaseChatModel heir returns.

model.invoke("Hello world")

AIMessage(content='Hello! How can I assist you today?', additional_kwargs={}, response_metadata={'model': 'llama3.1', 'created_at': '2026-01-20T14:41:12.602107097Z', 'done': True, 'done_reason': 'stop', 'total_duration': 5657624355, 'load_duration': 5473528405, 'prompt_eval_count': 12, 'prompt_eval_duration': 27060628, 'eval_count': 10, 'eval_duration': 140510213, 'logprobs': None, 'model_name': 'llama3.1', 'model_provider': 'ollama'}, id='lc_run--019bdbda-2b7d-7e22-afef-f1d811475446-0', tool_calls=[], invalid_tool_calls=[], usage_metadata={'input_tokens': 12, 'output_tokens': 10, 'total_tokens': 22})

Stream#

The stream method enables the outptu from the model to be obtained incrementally. It returns an AIMessageChunk containing the sequential output.

The following cell illustrates the process of gradual generation.

iterations = model.stream("What is the capital of France?")
for i in iterations:
    sleep(0.2)
    print(i.content.strip(), end="|")

The|capital|of|France|is|Paris|.|||

The message chunk example is shown in the following cell.

iterations = model.stream("What is the capital of GB?")
next(iterations)

AIMessageChunk(content='A', additional_kwargs={}, response_metadata={}, id='lc_run--019bdbe8-7ac6-7690-982c-81f9712d4061', tool_calls=[], invalid_tool_calls=[], tool_call_chunks=[])

Callbacks#

You can modify the behaviour of your Runnable using a callback. Extend the langchain_core.callbacks.BaseCallbackHandler with special methods and specify that Runnable have to use the callback handler during the infvocation in the “callbacks” attribute of the config.

The following table shows some of the methods that can be defined in the callback handler.

Method

Triggered When…

Useful For…

on_chain_start

A Chain (or Agent) begins.

Logging user inputs.

on_llm_start

The LLM API is about to be called.

Inspecting the exact prompt sent to the model.

on_llm_end

The LLM API returned the result.

Logging the information about model outputs.

on_llm_new_token

A token streams in (streaming only).

Real-time UI updates (typing effect).

on_tool_start

An Agent decides to use a tool.

Debugging which tools are being picked.

on_chain_error

An exception happens.

Capturing stack traces or alerting developers.

The following cell defines the MyCustomTracer, which extends the behaviour of the runnable associated with invoking LLMs.

from typing import Any, Dict, List
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult


class MyCustomTracer(BaseCallbackHandler):
    def on_llm_start(
        self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
    ) -> Any:
        print(f"\n[MY-TRACER] LLM Started with prompt: {prompts[0][:50]}...")

    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> Any:
        content = response.generations[0][0].text
        print(f"[MY-TRACER] LLM Finished. Output: {content[:50]}...")

The following cell illustrates the invocation of the model with attached callback.

ans = model.invoke(
    "Hello, I'm Fedor",
    config={"callbacks": [MyCustomTracer()]}
)

[MY-TRACER] LLM Started with prompt: Human: Hello, I'm Fedor...
[MY-TRACER] LLM Finished. Output: Nice to meet you, Fedor! Is there something I can ...
Contents