Multimodality & Voice

This content was generated with the assistance of AI and should be verified against the official documentation of each tool and framework before being relied on in production.

This section’s bibliography lists the reference material consulted while preparing these pages.

This page was written against langchain 1.4.3, langchain-core 1.6.5, langgraph 1.2.12, langchain-openai 1.6.6 and langchain-anthropic 1.7.4 on Python 3.12, with the openai SDK 3.20.0 for the speech examples, as verified on 2026-09-29. The running example is DocsAssistant, which answers questions over this site’s documentation using Ollama’s llama3.1:8b chat model and the nomic-embed-text embedding model over the pgvector store built in Integrating with LangChain. Ollama was not available while the snippets were checked, and no hosted provider was called: every output shown here was produced by running the code against the real packages with GenericFakeChatModel for the agent and an in-process HTTP transport that stands in for the provider’s servers. A real model would answer differently, and a real speech service returns real audio, but the request shapes, the block structure and the control flow are exactly what the libraries produce.

A text-only agent is a keyboard-shaped agent. Users have screenshots of an error, a PDF of the contract, a voice note recorded on a phone, and sometimes they want the answer spoken back. LangChain’s answer is small: media travels in the same messages list as text, as typed content blocks, and speech is handled by ordinary components placed before and after the agent. What models can do with pixels and waveforms, and why, is covered in Multimodal Models; this page is the plumbing. The full voice-agent design space — turn detection, barge-in, telephony, latency budgets — belongs to the Voice Agents sub-section (start with Voice Architectures); here it is limited to the pattern the official LangChain documentation builds.

Standard content blocks

Providers disagree on how to attach an image: an image_url part in one API, a source object in another. LangChain 1.x defines provider-neutral blocks in content_blocks (introduced in Chat Models & Messages) and translates them per provider:

Table 1. The media block types
type Carries Data forms

image

Photos, screenshots, diagrams

url, base64 plus mime_type (for example image/png), or a provider file_id

audio

Speech, recordings

url, base64 plus mime_type (for example audio/wav), or file_id

video

Clips

url, base64 plus mime_type (for example video/mp4), or file_id

file

PDFs and other documents

url, base64 plus mime_type (for example application/pdf), or file_id

text-plain

Text documents (.txt, .md) sent as a document, not as message text

Text, or url / base64

A base64 block must carry its mime_type. Extra provider-specific keys go at the top of the block or under extras; OpenAI, for example, wants a filename on a PDF. Not every model accepts every type, so check the provider’s integration page for supported formats and size limits.

DocsAssistant gets a new job: a user attaches an architecture screenshot, a page exported as PDF and a voice note, and asks whether they agree. A small helper turns a local file into the right block:

Turning local files into content blocks
import base64
import mimetypes
from pathlib import Path

from langchain.messages import HumanMessage


def file_block(path: str) -> dict:
    """Turn a local file into a standard LangChain content block."""
    mime, _ = mimetypes.guess_type(path)
    mime = {"audio/x-wav": "audio/wav"}.get(mime, mime)   # the name the provider APIs expect
    kind = {"image": "image", "audio": "audio", "video": "video"}.get((mime or "").split("/")[0], "file")
    block = {
        "type": kind,
        "base64": base64.b64encode(Path(path).read_bytes()).decode("ascii"),
        "mime_type": mime,
    }
    if kind == "file":
        block["filename"] = Path(path).name   # OpenAI requires a filename for PDFs
    return block


msg = HumanMessage(content_blocks=[
    {"type": "text", "text": "Is this diagram consistent with the attached page and the voice note?"},
    file_block("architecture.png"),
    file_block("vector-rag.pdf"),
    file_block("question.wav"),
])

for b in msg.content_blocks:
    print({k: (v if k != "base64" else f"<{len(v)} base64 chars>") for k, v in b.items()})

Output (executed, with a 64 by 32 pixel PNG, a one-page PDF and 0.1 s of silence as stand-ins):

{'type': 'text', 'text': 'Is this diagram consistent with the attached page and the voice note?'}
{'type': 'image', 'base64': '<224 base64 chars>', 'mime_type': 'image/png'}
{'type': 'file', 'base64': '<260 base64 chars>', 'mime_type': 'application/pdf', 'filename': 'vector-rag.pdf'}
{'type': 'audio', 'base64': '<4328 base64 chars>', 'mime_type': 'audio/wav'}

The 166-byte PNG became 224 base64 characters: base64 inflates media by about a third, and that inflated text lives in the message. Prefer a url block (or a provider file_id from a prior upload) for anything large, and think twice before letting a checkpointer (see Short-Term Memory & Checkpointers) store a base64 photo in every saved state. The block reference and the url, base64 and file_id forms are documented in LangChain — Messages.

Provider translation and capability checks

The standard block is what you write; the integration package rewrites it into the provider’s wire format when the request is made. This run points ChatOpenAI at an in-process transport that records the request body, then sends the same message to ChatAnthropic:

The same message, as OpenAI and Anthropic receive it
import json

import httpx
from langchain_anthropic import ChatAnthropic
from langchain_openai import ChatOpenAI

captured = {}


def fake_openai(request: httpx.Request) -> httpx.Response:
    captured["body"] = json.loads(request.content)      # keep what would have gone over the wire
    return httpx.Response(200, json={
        "id": "x", "object": "chat.completion", "created": 0, "model": "gpt-6-sol",
        "choices": [{"index": 0, "finish_reason": "stop",
                     "message": {"role": "assistant", "content": "It matches."}}],
    })


def shorten(x):
    if isinstance(x, dict):
        return {k: shorten(v) for k, v in x.items()}
    return x[:30] + "..." if isinstance(x, str) and len(x) > 60 else x


openai_model = ChatOpenAI(model="gpt-6-sol", api_key="unused",
                          http_client=httpx.Client(transport=httpx.MockTransport(fake_openai)))
print(openai_model.invoke([msg]).text)
for part in captured["body"]["messages"][0]["content"]:
    print(shorten(part))

claude = ChatAnthropic(model="claude-sonnet-5", api_key="unused")
try:
    claude.invoke([msg])
except ValueError as e:
    print("anthropic:", e)

Output (executed; the first line, the reply, is the canned response from the stand-in server, and the text part of the request is omitted here):

It matches.
{'type': 'image_url', 'image_url': {'url': 'data:image/png;base64,iVBORw0K...'}}
{'type': 'file', 'file': {'file_data': 'data:application/pdf;base64,JV...', 'filename': 'vector-rag.pdf'}}
{'type': 'input_audio', 'input_audio': {'data': 'UklGRqQMAABXQVZFZm10IBAAAAABAA...', 'format': 'wav'}}
anthropic: Block of type audio is not supported.

One neutral message became three OpenAI-specific shapes, and the filename requirement is why the helper adds filename to PDFs. The Anthropic integration accepted the image and the PDF (a separate run of the same message without the audio block got as far as opening a connection) and rejected the audio block before any network call. Two consequences follow. A model swap can silently change which modalities work, so gate on capability rather than trusting the block; and a rejected block is an exception in your code, not a model refusal you can prompt around. The fake HTTP transport does not check whether gpt-6-sol accepts audio input; consult the provider page for the model you actually deploy. Each provider’s integration page documents its own formats: ChatOpenAI and ChatAnthropic. Local models served through Ollama usually handle images only when the specific model is multimodal, and none of the examples here needs one; see Chat Models & Messages for ChatOllama.

Multimodal input to an agent

An agent from create_agent takes the same messages list, so attachments need no special handling: Agents with create_agent built the loop, and the message with media is just another HumanMessage in its state.

Passing the multimodal message to DocsAssistant
from langchain.agents import create_agent
from langchain.messages import AIMessage
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel


class FakeModel(GenericFakeChatModel):
    def bind_tools(self, tools, **kwargs):
        return self


agent = create_agent(
    FakeModel(messages=iter([AIMessage("The diagram matches the page.")])),
    tools=[],
    system_prompt="You are DocsAssistant. Compare attachments with the documentation.",
)
result = agent.invoke({"messages": [msg]})
print([m.type for m in result["messages"]])
print([b["type"] for b in result["messages"][0].content_blocks])
print(result["messages"][-1].text)

Output (executed):

['human', 'ai']
['text', 'image', 'file', 'audio']
The diagram matches the page.

Swap FakeModel for a real model that accepts all four block types and the agent can call docs_search between looking at the screenshot and answering. Everything in Tools applies unchanged; only the input is richer. Two habits keep multimodal agents cheap. Images and audio are billed in tokens and dominate the context window, so summarize or drop them from old turns with the techniques in Context Engineering for Agents. And treat text extracted from an image or PDF as untrusted input: an instruction embedded in a screenshot reaches the model exactly like one typed by the user, which is the subject of Guardrails & Security. See LangChain — Agents for the agent API.

Media coming back

Models can return media too. In LangChain the reply is an AIMessage whose content_blocks include image (or audio) blocks next to the text, with the bytes in base64. Reading them is the same loop for every provider that returns them:

Saving a generated image from an AIMessage
import base64
from pathlib import Path

from langchain.messages import AIMessage

png = Path("architecture.png").read_bytes()
reply = AIMessage(content_blocks=[
    {"type": "text", "text": "Here is the diagram."},
    {"type": "image", "base64": base64.b64encode(png).decode(), "mime_type": "image/png"},
])

for block in reply.content_blocks:
    if block["type"] == "image":
        Path("generated.png").write_bytes(base64.b64decode(block["base64"]))
        print("saved", block["mime_type"], len(png), "bytes")
print(reply.text)

Output (executed, on a hand-built reply rather than a model call):

saved image/png 166 bytes
Here is the diagram.

How to ask a provider for an image in the first place is provider-specific; the OpenAI integration page documents image generation as a built-in tool the model may call. See ChatOpenAI for that and for the native-format alternatives to the standard blocks.

Transcription and text-to-speech

Most production voice systems do not send audio to the chat model at all. They use dedicated speech-to-text (STT) and text-to-speech (TTS) services around a text model, which keeps the strongest reasoning model in charge and makes each stage replaceable. LangChain does not wrap these services behind one interface; the official voice-agent guide writes small adapters around the vendors' SDKs and WebSocket APIs (AssemblyAI for STT and Cartesia for TTS) and says the same shape works for other providers. This page uses the openai SDK because it covers both directions in one package:

Transcription and speech synthesis with the OpenAI SDK
import httpx
from openai import OpenAI

seen = []


def fake_openai(request: httpx.Request) -> httpx.Response:
    seen.append((request.method, request.url.path, request.headers["content-type"].split(";")[0]))
    if request.url.path.endswith("/transcriptions"):
        return httpx.Response(200, json={"text": "How do I create an HNSW index?"})
    return httpx.Response(200, content=b"RIFF....WAVEfmt fake-audio", headers={"content-type": "audio/wav"})


client = OpenAI(api_key="unused", http_client=httpx.Client(transport=httpx.MockTransport(fake_openai)))
# Real code: client = OpenAI()  -- the key is read from the OPENAI_API_KEY environment variable.


def transcribe(path: str) -> str:
    with open(path, "rb") as f:
        return client.audio.transcriptions.create(model="gpt-4o-transcribe", file=f).text


def speak(text: str) -> bytes:
    return client.audio.speech.create(
        model="gpt-4o-mini-tts", voice="alloy", input=text, response_format="wav",
        instructions="Calm, concise support agent.",
    ).read()


print(transcribe("question.wav"))
print(speak("Use CREATE INDEX ... USING hnsw."))
print(seen)

Output (executed against the stand-in transport, so the audio bytes are fake):

How do I create an HNSW index?
b'RIFF....WAVEfmt fake-audio'
[('POST', '/v1/audio/transcriptions', 'multipart/form-data'), ('POST', '/v1/audio/speech', 'application/json')]

The model names come from the SDK’s own list of transcription and speech models in version 3.20.0. The two endpoints differ in shape: transcription uploads the audio as multipart/form-data and returns text; speech posts JSON and returns audio bytes, and its instructions parameter steers delivery. Keys come from OPENAI_API_KEY in real code, never from source. Other providers offer the same two operations under different names, and a self-hosted server that speaks the same audio API can be reached by pointing the client at its base URL, which keeps the audio on your own infrastructure. Look up the request formats in OpenAI — Speech to text and OpenAI — Text to speech.

A model that accepts an audio content block, as in the first section, is a different design: the model listens directly and can use tone and pauses, at the price of a narrower choice of models and less visibility into what it understood. Transcribing first gives you a text you can log, evaluate and guard.

The STT, agent, TTS sandwich

The official guide, Build a voice agent with LangChain, names two architectures. In the sandwich, speech-to-text feeds a text agent whose reply goes through text-to-speech; in speech-to-speech, one multimodal model consumes and produces audio natively. The sandwich is the one the guide builds, because it keeps full control of each part, gives access to the newest text models and tools, and has clear boundaries between components. Its costs are more services to run and the loss of tone and emotion when speech becomes text.

flowchart LR U([Caller audio]) --> STT[Speech-to-text
partial and final transcripts] STT -- final transcript --> A[create_agent
tools and checkpointer] A -- reply tokens --> S[Sentence buffer] S --> TTS[Text-to-speech
audio chunks] TTS --> P([Playback]) A <-. thread_id .-> M[(Checkpointer)]
Figure 1. The sandwich architecture

The diagram above is the whole design. Each box is independent, so an STT vendor, a TTS voice or the chat model can change without touching the other two, and the agent in the middle is exactly the one built in Agents with create_agent, with the checkpointer from Short-Term Memory & Checkpointers keeping the conversation across spoken turns.

Batch turn with LCEL

The simplest sandwich handles one recorded utterance and returns one audio reply. Because each stage is a function, the whole thing is an LCEL chain (see LCEL & Runnables), and naming the stages with with_config(run_name=…​) makes them show up as separate steps in a trace:

A batch sandwich as an LCEL chain
from langchain.agents import create_agent
from langchain.messages import AIMessage, HumanMessage
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.runnables import RunnableLambda


class FakeModel(GenericFakeChatModel):
    def bind_tools(self, tools, **kwargs):
        return self


agent = create_agent(FakeModel(messages=iter([AIMessage("Create it with CREATE INDEX USING hnsw.")])),
                     tools=[], system_prompt="Answer in one short spoken sentence.")

sandwich = (
    RunnableLambda(transcribe).with_config(run_name="stt")
    | RunnableLambda(lambda q: {"messages": [HumanMessage(q)]})
    | agent
    | RunnableLambda(lambda state: state["messages"][-1].text).with_config(run_name="reply_text")
    | RunnableLambda(speak).with_config(run_name="tts")
)
audio = sandwich.invoke("question.wav")
print(type(audio).__name__, len(audio), "bytes")

Output (executed with the transcribe and speak functions from the previous section):

bytes 26 bytes

The 26 bytes are the fake audio from the stand-in transport. This version is easy to test and trace, but the user hears nothing until the whole reply exists, which for a voice assistant feels like a hung phone line. See LangChain — Build a voice agent with LangChain.

Streaming turn

The official guide streams at every stage so that speech starts before the agent has finished writing. The agent stage uses astream_events(…​, version="v3") and iterates stream.messages and each message’s text, which yields tokens as they are generated. The same shape works here, with one addition: tokens are grouped into sentences before they reach TTS, because synthesizing single words sounds choppy and per-request overhead would dominate.

A streaming voice turn with sentence buffering and a shared thread
import asyncio
import re
from collections.abc import AsyncIterator

from langchain.agents import create_agent
from langchain.messages import AIMessage, HumanMessage
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langgraph.checkpoint.memory import InMemorySaver

VOICE_PROMPT = (
    "You are DocsAssistant answering by voice. Reply in one or two short sentences of plain text: "
    "no markdown, lists, code or emojis, because a speech engine reads your answer aloud."
)


class FakeModel(GenericFakeChatModel):
    def bind_tools(self, tools, **kwargs):
        return self


model = FakeModel(messages=iter([
    AIMessage("Create it with CREATE INDEX USING hnsw. Raise ef_search when recall drops."),
    AIMessage("Yes, IVFFlat is the alternative."),
]))
agent = create_agent(model, tools=[], system_prompt=VOICE_PROMPT, checkpointer=InMemorySaver())

SENTENCE_END = re.compile(r"(?<=[.!?])\s+")


async def sentences(tokens: AsyncIterator[str]) -> AsyncIterator[str]:
    """Group streamed tokens into sentences so TTS gets natural units, not single words."""
    buffer = ""
    async for token in tokens:
        buffer += token
        *done, buffer = SENTENCE_END.split(buffer)
        for sentence in done:
            yield sentence
    if buffer.strip():
        yield buffer.strip()


async def voice_turn(audio_path: str, thread_id: str) -> list[bytes]:
    question = await asyncio.to_thread(transcribe, audio_path)            # STT
    print("heard:", question)
    stream = await agent.astream_events(                                   # agent
        {"messages": [HumanMessage(question)]},
        {"configurable": {"thread_id": thread_id}},
        version="v3",
    )

    async def tokens() -> AsyncIterator[str]:
        async for message in stream.messages:
            async for token in message.text:
                yield token

    clips = []
    async for sentence in sentences(tokens()):                             # TTS, one sentence at a time
        print("speaking:", sentence)
        clips.append(await asyncio.to_thread(speak, sentence))
    return clips


async def main() -> None:
    for _ in range(2):    # two turns on one thread: the checkpointer carries the conversation
        clips = await voice_turn("question.wav", "voice-thread-1")
        print(len(clips), "audio clips")
    state = await agent.aget_state({"configurable": {"thread_id": "voice-thread-1"}})
    print(len(state.values["messages"]), "messages in the thread")


asyncio.run(main())

Output (executed; the first reply has two sentences, so two audio clips, and the second turn reuses the thread):

heard: How do I create an HNSW index?
speaking: Create it with CREATE INDEX USING hnsw.
speaking: Raise ef_search when recall drops.
2 audio clips
heard: How do I create an HNSW index?
speaking: Yes, IVFFlat is the alternative.
1 audio clips
4 messages in the thread

The blocking SDK calls run in asyncio.to_thread so they do not stall the event loop; a vendor with a streaming WebSocket API, as in the official guide, would replace them with a producer that sends text as it arrives and a consumer that yields audio chunks. LangChain marks the v3 streaming protocol as experimental, so expect LangChainBetaWarning on the two calls above and check the streaming page before upgrading; the more stable stream modes are covered in Streaming. The demo repository referenced by the guide, voice-sandwich-demo, shows the full pipeline with a browser client and WebSockets.

What the pattern needs beyond the code

  • Write for the ear. The system prompt above forbids markdown, lists and code, since the speech engine would read the symbols aloud. Keep answers to a sentence or two, and make DocsAssistant offer to send the long version to the user’s screen instead of reading a table.

  • Latency is the product. Time to first audio is STT finalization plus time to the first sentence plus TTS start. Stream every stage, stream the model, keep the tool loop short (see Tool Design), and use the smaller models for routine turns.

  • Interruptions. Real callers talk over the agent. That needs turn detection, cancelling the TTS clips still queued and discarding the rest of the reply; it is a property of the audio transport rather than of LangChain, and is where a dedicated voice framework earns its keep.

  • Approvals. A destructive tool, such as DocsAssistant opening a GitHub issue, still needs a human decision. A spoken "yes" is weak evidence; confirm on a screen or read the action back and require an explicit phrase. See Human-in-the-Loop & Time Travel.

  • Privacy. Voice is personal data, and transcripts and recordings sent to a hosted STT or TTS service leave your infrastructure. Choose retention settings deliberately, and see Agent Guardrails & Safety and Guardrails & Security.

  • Tracing. LangSmith recommends recording a whole voice conversation as a single trace with a combined audio file on the root run, and marking it as audio, so barge-ins and gaps between speakers stay visible; see LangSmith — Voice tracing fundamentals and Observability with LangSmith.

Choosing an approach

Table 2. Media and voice options
Need Use Trade-off

Answer questions about an image or PDF

image or file blocks on a vision-capable chat model

Media tokens are expensive; check per-model limits

The user speaks and reads text back

STT before the agent

Simple, testable; loses tone

The user speaks and hears the answer

The sandwich: STT, agent, TTS with sentence streaming

Several services and a latency budget to manage

Audio in, audio out with the fewest parts

A speech-to-speech model

Fewer model choices, less control and visibility

Phone calls, barge-in, telephony

A voice framework; see LiveKit Agents or Pipecat

Outside LangChain proper

Multimodal attachments and voice both widen the attack surface and the bill, so build the text agent first, evaluate it as in Agent Evaluation, and only then put media around it. For the models that can run on your own machine, see Running LLMs Locally.

References