LangChain

OverviewCopy Anchor Link

LangchainProcessor puts a LangChain.js runnable where an LLM service would sit in the pipeline. It takes the latest user message from the context, streams it through the runnable and pushes the output as text frames.

• LangchainProcessor: a frame processor that wraps any Runnable from @langchain/core: a prompt piped into a chat model, a chain with message history, or a compiled LangGraph graph.

InstallationCopy Anchor Link

@langchain/core is a peer dependency (^1.0.0): @pipedog/langchain imports from it but does not install it, so you pick the version. Add the chat model integration you use as well.

bun add @pipedog/langchain @langchain/core @langchain/openai

import { LangchainProcessor } from "@pipedog/langchain/langchain";

PrerequisitesCopy Anchor Link

• Whatever your runnable needs. With ChatOpenAI that is an OpenAI API key in OPENAI_API_KEY.
• The processor itself needs no key.

LangchainProcessorCopy Anchor Link

ConfigurationCopy Anchor Link

The constructor is new LangchainProcessor(chain, options). chain is a Runnable from @langchain/core/runnables. The processor calls only its stream(input, config) method.

What the runnable receives:

• input: { [transcriptKey]: text }, where text is the trimmed content of the last context message. With the default key that is { input: "..." }, so a prompt template should use an {input} variable.
• config: { configurable: { session_id }, signal }. session_id is the value passed to setParticipantId() (null until then), which is what RunnableWithMessageHistory reads. signal aborts the stream when the pipeline is interrupted or cancelled.

What it must yield: strings or AIMessageChunks. A chunk's string content becomes the frame text. List content (multimodal) is turned into text with JSON.stringify. Any other value becomes an empty TextFrame.

UsageCopy Anchor Link

Build the runnable with LangChain:

import { ChatPromptTemplate } from "@langchain/core/prompts";
import { ChatOpenAI } from "@langchain/openai";

const prompt = ChatPromptTemplate.fromMessages([
  ["system", "You are a helpful voice assistant. Keep answers short."],
  ["human", "{input}"],
]);

export const chain = prompt.pipe(new ChatOpenAI({ model: "gpt-4o-mini" }));

Then wrap it in the processor, in place of an LLM service:

import { LangchainProcessor } from "@pipedog/langchain/langchain";

// Any @langchain/core Runnable: a prompt piped into a chat model, a compiled LangGraph, ...
declare const chain: ConstructorParameters<typeof LangchainProcessor>[0];

const langchain = new LangchainProcessor(chain, { transcriptKey: "input" });

// Sent to the chain as `config.configurable.session_id`, for message history.
langchain.setParticipantId("participant-123");

NotesCopy Anchor Link

• Consumes LLMContextFrame. Every other frame passes through.
• The last message must be a standard message with string content. A provider-specific message (LLMSpecificMessage) or list content is not sent to the chain: the context frame is passed on instead.
• For each run it pushes LLMFullResponseStartFrame, one TextFrame per streamed chunk (with includesInterFrameSpaces set) and LLMFullResponseEndFrame. The end frame is pushed even when the chain throws.
• A chain error is pushed upstream as an ErrorFrame (Unknown error occurred: ...).
• A compiled LangGraph graph is a Runnable, so it works here too. Its default stream() yields state updates, not text, so wrap it (for example in a RunnableLambda that returns the final message) or stream with streamMode: "messages". The package tests show both.

Runtime notesCopy Anchor Link

Generated from pipecat.runtimes and pipecat.compat in packages/langchain/package.json: ✅ means the compat smoke test passes on that runtime.

• Each step of the stream is a cancellation point, so an interruption stops the chain mid-stream. The iterator is closed in a finally block.
• Upstream logs a warning on Python's GeneratorExit. That branch has no JS equivalent and is not ported.
• The package targets Bun, Node, Deno and Cloudflare Workers, and uses no runtime-specific APIs. Whether a chain runs on a given runtime depends on the LangChain integrations it uses.