Models, messages and providers

init_chat_model, provider packages, message types, streaming, and swapping providers without rewriting the rest of the application.

Initialising a model

import os
from langchain.chat_models import init_chat_model

# one factory, many providers: the model string is parsed into provider + name
model = init_chat_model("gpt-4o-mini", model_provider="openai", temperature=0)
claude = init_chat_model("claude-3-5-sonnet-latest", model_provider="anthropic")
local = init_chat_model("llama3.2", model_provider="ollama", base_url="http://localhost:11434")

print(model.model_name if hasattr(model, "model_name") else type(model).__name__)

# provider packages are separate installs; a missing one fails at import time
# pip install langchain-openai langchain-anthropic langchain-ollama

response = model.invoke("Name two differences between TCP and UDP.")
print(response.content)
print(response.usage_metadata)

• temperature=0 for extraction and classification; higher values for brainstorming. Determinism is a per-call property, so set it where you create the model.
• Every provider returns a slightly different object, but invoke, batch, stream and ainvoke are the same interface, which is the point of the abstraction.
• usage_metadata carries input and output token counts when the provider reports them. Log it — it is your only per-call cost signal.
• Keep the model configuration in one place (a factory function or a config file) so a provider swap is a one-line change rather than a search-and-replace.

Message types

from langchain_core.messages import (
    SystemMessage, HumanMessage, AIMessage, ToolMessage, trim_messages
)

messages = [
    SystemMessage("You answer in British English, in at most three sentences."),
    HumanMessage("Summarise the point of idempotency in APIs."),
    AIMessage("An idempotent request can be repeated without changing the outcome."),
    HumanMessage("Give one concrete example."),
]

# a tool result is a distinct message type tied to a tool call id
tool_result = ToolMessage(content="charge_42 captured", tool_call_id="call_abc123")
print(tool_result.type)

# trim a long history by tokens, keeping the system message and the latest turn
trimmed = trim_messages(
    messages,
    max_tokens=200,
    strategy="last",
    token_counter=model,
    include_system=True,
    start_on="human",
    allow_partial=False,
)
print([m.type for m in trimmed])

Streaming and batching

# streaming: print tokens as they arrive instead of waiting for the whole reply
for chunk in model.stream("Write a haiku about deployment."):
    print(chunk.content, end="", flush=True)

# async streaming for a web endpoint
import asyncio

async def stream_reply(prompt):
    pieces = []
    async for chunk in model.astream(prompt):
        pieces.append(chunk.content)
        yield chunk.content
    print("\nfull length:", len("".join(pieces)))

# batching: parallel calls through the provider's concurrency, not a Python loop
batch_inputs = [
    "Classify as positive or negative: the build passed first time.",
    "Classify as positive or negative: the deploy failed again.",
    "Classify as positive or negative: latency improved by 40%.",
]
results = model.batch(batch_inputs, config={"max_concurrency": 4})
for text, result in zip(batch_inputs, results):
    print(text.split(":")[1], "->", result.content.strip())

• Streaming does not reduce total generation time, it reduces perceived latency. Any user-facing path should stream.
• Provider SDKs retry transient failures internally; LangChain surfaces persistent errors. Catch them at the boundary of your application, not inside a chain.
• batch with max_concurrency is the correct way to run many independent prompts. A Python for loop with invoke is serial and wastes most of your rate limit.
• A streaming response may arrive in chunks that split a JSON object. If you parse streaming output, buffer until the delimiter appears rather than parsing every chunk.

FAQ

Related

Prompt templates and chains Structured output and output parsers

Last refreshed 2026-09-18.