Optimizing Agent Cost & Latency in Practice (LangChain, LangGraph, AutoGen Versions)

Overview

Splitting an agent into a fast planner and a slow executor is a decision you make once, on paper. Wiring it into a real codebase is a decision you make three times, differently, depending on whether you're holding a LangChain pipeline, a LangGraph state machine, or a team of AutoGen agents talking to each other. The pattern is identical. The code that expresses it is not, and we think the gap between the two is where most of the actual engineering time goes.

Most naive agents do the expensive thing every time, regardless of framework: the full conversation and every retrieved document go to the most capable, most costly model on every turn, whether or not the turn warrants it. A planning step that just needs to enumerate three subtasks pays frontier-model prices. A plain greeting triggers a vector search and a context-stuffed generation. Fixing that is mostly a matter of doing the cheap thing first and the expensive thing only when it earns its keep, and that holds regardless of which framework you're in.

Agent workloads split cleanly into two kinds of work: heavyweight thinking (planning, decomposition, global reasoning) and high-stakes execution (final answers, tool actions, anything user-facing). Map those onto two model tiers, cheap and accurate, and the cheap tier doesn't need to be perfect. It only needs to be good enough to plan, triage, and compress, which are tasks where a smaller model is often surprisingly competitive.

LangChain: Fast Planner + Slow Executor + Caching + Context Optimization

LangChain's strength here is composition. Every model, prompt, and helper is a Runnable, so you pipe them together with the | operator and swap pieces in and out without rewiring the whole agent. Build the flow one Runnable at a time, then compose.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_core.output_parsers import StrOutputParser

# Fast, cheap model (planning, summaries, compression)
fast_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Slow, high-quality model (final execution, user-visible answers)
slow_llm = ChatOpenAI(model="gpt-4.1", temperature=0)

Prompt templates now live in langchain_core.prompts rather than the older langchain.prompts path. Keeping imports on the langchain_core namespace is the safest choice across recent versions.

The planner only enumerates steps. It never produces the final answer, which is exactly why it belongs on the cheap model.

planner_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a planner. Break the user task into clear steps. No final answer."),
    ("user", "{question}")
])

planner_chain = planner_prompt | fast_llm | StrOutputParser()

Lazy retrieval and context compression assume you already have a retriever, such as a vector store:

from langchain_core.documents import Document

def compress_docs(docs: list[Document]) -> str:
    if not docs:
        return ""
    joined = "\n\n".join(d.page_content for d in docs)
    prompt = f"Summarize the following context in <= 5 bullet points:\n\n{joined}"
    return (fast_llm | StrOutputParser()).invoke(prompt)

def maybe_retrieve(question: str, retriever) -> list[Document]:
    judge_prompt = f"""
    Question: {question}
    Do we need to look up external documents to answer this?
    Respond with a single word: yes or no.
    """
    resp: str = (fast_llm | StrOutputParser()).invoke(judge_prompt)
    if "yes" in resp.lower():
        return retriever.invoke(question)
    return []

The maybe_retrieve guard turns retrieval into a decision instead of a reflex. A one-token yes/no from the cheap model is far cheaper than an embedding query plus a context-stuffed generation, and on plenty of turns the honest answer is no. compress_docs then shrinks whatever you do retrieve into a handful of bullets, so the expensive model reads a paragraph instead of a dossier. Retrievers are Runnables now, so call retriever.invoke(question) rather than the deprecated get_relevant_documents.

Wrapped as a Runnable:

def lazy_context_fn(inputs: dict):
    question = inputs["question"]
    retriever = inputs["retriever"]
    docs = maybe_retrieve(question, retriever)
    compressed = compress_docs(docs)
    return {"compressed_context": compressed}

lazy_context = RunnableLambda(lazy_context_fn)

By the time the slow model runs, the hard thinking is done: it has a plan to follow and a compressed context to ground it. That means the expensive call is short, focused, and predictable, which is exactly when a frontier model is worth paying for.

executor_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a precise, careful assistant. Follow the plan and use context."),
    ("user", "Question: {question}\n\nPlan:\n{plan}\n\nContext:\n{compressed_context}")
])

executor_chain = executor_prompt | slow_llm | StrOutputParser()

LangChain ships built-in caching too, with a global LLM cache you can set via set_llm_cache to deduplicate identical model calls. Here we keep it explicit and memoize just the planner step, so repeated questions skip planning entirely.

from functools import lru_cache

@lru_cache(maxsize=1024)
def cached_plan(question: str) -> str:
    return planner_chain.invoke({"question": question})

def optimized_agent_lc(question: str, retriever):
    plan = cached_plan(question)
    ctx = lazy_context.invoke({"question": question, "retriever": retriever})
    answer = executor_chain.invoke({
        "question": question,
        "plan": plan,
        "compressed_context": ctx["compressed_context"],
    })
    return {"plan": plan, "answer": answer}

One caveat worth flagging: lru_cache keys on the exact question string, so it only helps with verbatim repeats. For production you'd usually want a semantic cache (embed the question, match on similarity) or LangChain's own cache backends, which persist across processes. The principle holds either way: don't re-plan what you've already planned.

That gives you fast, cached planning, lazy retrieval, compressed context, and a slow, accurate executor, all inside a handful of Runnables.

LangGraph: Routing by Stage and Optimizing Execution Paths

The LangChain version above is linear: plan, maybe-retrieve, execute, done. That's fine until you want to branch, retry on a weak answer, ask for clarification, or skip a stage entirely. LangGraph is built for exactly that control flow, modeling the agent as an explicit state machine where each node reads and updates a shared state object.

from typing import TypedDict, Optional, Any

class AgentState(TypedDict):
    question: str
    plan: Optional[str]
    compressed_context: Optional[str]
    answer: Optional[str]
    confidence: Optional[float]
    retriever: Any  # store a handle/identifier

Use Any, capital A, imported from typing, rather than the lowercase builtin any, which is a function and not a type. Each node below is a plain function that takes the state and returns the keys it wants to update; LangGraph merges those back in, so a node only has to return what it changes.

def planner_node(state: AgentState) -> AgentState:
    plan = planner_chain.invoke({"question": state["question"]})
    return {**state, "plan": plan}

def context_node(state: AgentState) -> AgentState:
    docs = maybe_retrieve(state["question"], state["retriever"])
    compressed = compress_docs(docs)
    return {**state, "compressed_context": compressed}

The executor node estimates confidence inline:

import json

def execution_node(state: AgentState) -> AgentState:
    prompt = f"""
    Question: {state['question']}
    Plan:
    {state['plan']}
    Context:
    {state['compressed_context']}

    Respond JSON:
    {{"answer": "<answer>", "confidence": 0-1}}
    """
    resp = (slow_llm | StrOutputParser()).invoke(prompt)
    data = json.loads(resp)
    return {**state, "answer": data["answer"], "confidence": data["confidence"]}

Asking the model to self-report a confidence score in the same call is cheap, but free-form JSON parsing is brittle. A stray code fence or trailing comment breaks json.loads. In production, prefer structured outputs: bind a schema with slow_llm.with_structured_output(...), or a library like Instructor, so the model is constrained to valid JSON and the manual parse goes away.

The router is where the graph earns its cost. The expensive executor runs once, and a cheap routing decision determines whether you stop, ask for clarification, or retry:

def router_node(state: AgentState) -> str:
    conf = state.get("confidence") or 0.0
    if conf > 0.8:
        return "done"
    if 0.5 < conf <= 0.8:
        return "clarify"
    return "retry"

from langgraph.graph import StateGraph, START, END

workflow = StateGraph(AgentState)
workflow.add_node("plan", planner_node)
workflow.add_node("context", context_node)
workflow.add_node("execute", execution_node)
workflow.add_node("clarify", lambda s: {**s, "answer": "I need clarification."})
workflow.add_node("retry", planner_node)  # trivial retry; customize as needed

workflow.add_edge(START, "plan")
workflow.add_edge("plan", "context")
workflow.add_edge("context", "execute")
workflow.add_conditional_edges(
    "execute",
    router_node,
    {"done": END, "clarify": "clarify", "retry": "retry"},
)

graph = workflow.compile()

done |-- 0.5 to 0.8 --> clarify node --> done |-- below 0.5 --> retry --> back to plan node -->

Figure 1 — The LangGraph router. The expensive executor runs exactly once per pass; confidence decides whether the cheap path (stop, or ask one clarifying question) is enough, or whether the graph loops back.

add_edge(START, "plan") is the current idiom for the entry point; the older set_entry_point still works but the explicit START node reads more clearly next to END.

LangGraph now encodes model mixing, lazy context, and confidence-based routing as a proper state machine, with room to add whatever retry or repair logic your task actually needs.

AutoGen: Planner-Executor Multi-Agent with Cost/Latency Optimization

AutoGen is built for multi-agent setups, which map naturally onto a planner agent (fast model), an executor agent (slow model), and an optional critic agent (fast model, for validation).

A note on versions before diving in. AutoGen has two API generations. The original conversational API, from autogen import AssistantAgent with a dict llm_config and synchronous generate_reply, is shown below because it's the most concise way to express the pattern, and it's still maintained, now also under the community AG2 fork. The newer Microsoft autogen-agentchat package replaces this with a per-agent model_client (such as OpenAIChatCompletionClient) and async methods like on_messages and run. The optimization pattern is identical in both; only the call surface differs.

from autogen import AssistantAgent, UserProxyAgent

planner = AssistantAgent(
    name="planner",
    system_message="You are a planner. Break tasks into steps. No final answers.",
    llm_config={"model": "gpt-4o-mini", "temperature": 0},
)

executor = AssistantAgent(
    name="executor",
    system_message="You execute plans precisely and safely.",
    llm_config={"model": "gpt-4.1", "temperature": 0},
)

critic = AssistantAgent(
    name="critic",
    system_message="You check answer quality and rate confidence 0-1.",
    llm_config={"model": "gpt-4o-mini", "temperature": 0},
)

user = UserProxyAgent(name="user")

The cost lever sits in those llm_config blocks: the planner and critic run on the cheap model, and only the executor pays for the frontier one. Two of the three agents on this team are inexpensive.

def lazy_rag(question, retriever):
    docs = maybe_retrieve(question, retriever)
    return compress_docs(docs)

Orchestrating planner, executor, and critic by hand, rather than handing them to a free-flowing group chat, is deliberate. A group chat can balloon token usage as agents talk past each other, so for a cost-sensitive pipeline an explicit call order is usually cheaper and more predictable:

import json

def optimized_autogen_agent(question: str, retriever):
    plan = planner.generate_reply(messages=[{"role": "user", "content": question}])

    compressed_context = lazy_rag(question, retriever)

    exec_prompt = f"""
    Question: {question}
    Plan:
    {plan}
    Compressed context:
    {compressed_context}

    Provide a JSON object with:
    - answer: string
    """
    exec_msg = executor.generate_reply(messages=[{"role": "user", "content": exec_prompt}])
    answer = json.loads(exec_msg)["answer"]

    critic_prompt = f"""
    Evaluate the following answer for factual accuracy and appropriateness.
    Question: {question}
    Answer: {answer}

    Respond JSON:
    {{"confidence": 0-1, "comment": "..."}}
    """
    critic_msg = critic.generate_reply(messages=[{"role": "user", "content": critic_prompt}])
    critic_data = json.loads(critic_msg)

    return {
        "plan": plan,
        "answer": answer,
        "confidence": critic_data["confidence"],
        "critic_comment": critic_data["comment"],
    }

The same parsing caveat from the LangGraph section applies: guard the json.loads calls, or wrap them in a structured-output helper, since a stray character in the model's reply otherwise takes down the whole call. This version gives you a fast planner agent, a slow executor agent, a fast critic for confidence, and lazy RAG with compressed context, and you can extend it with AutoGen's own tools mechanism to expose the retriever, the compressor, or per-agent logging as first-class tools.

Measuring the Wins

Optimizations you can't measure tend to quietly regress. Before tuning anything, capture a baseline, so you know what you actually saved and so the next person who touches the agent can tell whether their change helped or hurt.

Three numbers matter most: total tokens, split into prompt and completion since they price differently; wall-clock latency per request; and the fraction of requests that hit the expensive model versus the cheap one. A lightweight wrapper around each model call is enough to start attributing cost to a stage, planning, retrieval, execution, or critique, so you can see where the money actually goes.

import time
from collections import defaultdict

stats = defaultdict(lambda: {"calls": 0, "prompt_tokens": 0, "completion_tokens": 0, "seconds": 0.0})

def track(stage: str, llm, prompt: str) -> str:
    start = time.perf_counter()
    resp = llm.invoke(prompt)
    elapsed = time.perf_counter() - start

    usage = resp.response_metadata.get("token_usage", {})
    s = stats[stage]
    s["calls"] += 1
    s["prompt_tokens"] += usage.get("prompt_tokens", 0)
    s["completion_tokens"] += usage.get("completion_tokens", 0)
    s["seconds"] += elapsed
    return resp.content

For anything beyond a quick experiment, lean on real tracing. LangChain and LangGraph emit traces to LangSmith out of the box, and OpenTelemetry gives you a vendor-neutral way to ship spans for every model and tool call into whatever observability backend you already run. The instrumentation is the same regardless of framework; what you're watching for is always the same pair of curves, cost per request trending down and the expensive-model hit rate staying low without quality falling off.

Which Framework Should You Use?

All three implement the same playbook: fast planner, slow executor, lazy retrieval, compressed context, confidence routing. The choice is mostly about the shape of your problem, not which one runs faster.

LangChain suits composable chains, a simple architecture, and quick iteration. LangGraph earns its keep once you need real control flow: retries, fallback routing, production stability under a state machine you can reason about. AutoGen fits multi-agent collaboration, planner-executor-critic patterns, and tool-agent ecosystems where the problem genuinely is a team of specialists rather than a single pipeline with stages.

We'd point most teams at LangChain first, to validate the pattern quickly, then graduate the same flow into LangGraph once retries, branching, and durable state actually matter. The planner, executor, and helpers carry over almost unchanged. Reach for AutoGen when the collaboration is the point, not an afterthought bolted onto a pipeline.

Conclusion

Three frameworks, one pattern. LangChain expresses it as piped Runnables, LangGraph as an explicit graph with a router, AutoGen as a small team where only one member is expensive. Whichever you pick, the code that saves money is doing the same job underneath: deciding what actually needs the frontier model, and skipping or reusing everything else.

None of it is worth wiring in blind. Track the expensive-model hit rate from day one, hold an evaluation set fixed while you turn each pattern on, and let the numbers from your own pipeline, not a framework's default config, tell you when you're done.