headroomlabs-ai/headroom · error · ImportError

LangChain is required for this integration. Install with: pi

Error message

LangChain is required for this integration. Install with: pip install headroom[langchain] or: pip install langchain-core

What it means

Raised by _check_langchain_available() in the LangChain streaming integration when langchain-core is missing (AIMessageChunk and ChatGenerationChunk failed to import and were stubbed to `object`). It guards the streaming wrapper that produces StreamingMetrics (output_tokens, chunk_count) around token streams from an OpenAIProvider.

Source

Thrown at headroom/integrations/langchain/streaming.py:46

try:
    from langchain_core.messages import AIMessageChunk
    from langchain_core.outputs import ChatGenerationChunk

    LANGCHAIN_AVAILABLE = True
except ImportError:
    LANGCHAIN_AVAILABLE = False
    AIMessageChunk = object  # type: ignore[misc,assignment]
    ChatGenerationChunk = object  # type: ignore[misc,assignment]

from headroom.providers import OpenAIProvider

logger = logging.getLogger(__name__)


def _check_langchain_available() -> None:
    """Raise ImportError if LangChain is not installed."""
    if not LANGCHAIN_AVAILABLE:
        raise ImportError(
            "LangChain is required for this integration. "
            "Install with: pip install headroom[langchain] "
            "or: pip install langchain-core"
        )


@dataclass
class StreamingMetrics:
    """Metrics from a streaming response."""

    output_tokens: int
    chunk_count: int
    content_length: int
    start_time: datetime
    end_time: datetime | None
    duration_ms: float | None

    def to_dict(self) -> dict[str, Any]:

View on GitHub (pinned to 322425c43b)

Solutions

  1. Install `pip install 'headroom[langchain]'`
  2. If you installed it mid-session in a notebook, restart the kernel/process — LANGCHAIN_AVAILABLE is evaluated once at import time
  3. Verify with `python -c "from headroom.integrations.langchain import streaming"`

Example fix

# before
stream = headroom_stream(provider, messages)  # ImportError raised by guard

# after
# pip install 'headroom[langchain]'; restart process
stream = headroom_stream(provider, messages)
Defensive patterns

Strategy: validation

Validate before calling

import importlib.util
assert importlib.util.find_spec('langchain_core'), 'langchain-core required for the streaming integration'

Type guard

def streaming_wrapper_ready() -> bool:
    from headroom.integrations.langchain import streaming
    return streaming.LANGCHAIN_AVAILABLE

Try / catch

try:
    for chunk in headroom_stream(...):
        yield chunk
except ImportError as e:
    if 'langchain' in str(e):
        yield from raw_stream(...)  # bypass the LangChain chunk wrappers
    else:
        raise

Prevention

When it happens

Trigger: Starting or consuming a headroom-wrapped streaming response (any API on this module that builds AIMessageChunk/ChatGenerationChunk objects) when langchain-core is absent.

Common situations: An app that previously used non-streaming headroom calls adds streaming via the LangChain integration without adding the dependency; notebooks where the kernel was started before installing langchain-core (module cached with LANGCHAIN_AVAILABLE=False — needs kernel restart even after pip install).

Related errors

AI-assisted analysis of headroomlabs-ai/headroom@322425c43b (2026-08-15). Data as JSON: /api/errors/b9014145b61b3378. Report an issue: GitHub.