advanced⏱35 minutesLição 9 de 10

Otimização

Otimize aplicações LangGraph — reduza chamadas de nós, estratégias de cache, lotação de requisições LLM, gerenciamento de tokens e ajuste de desempenho do grafo.

Otimização

As LangGraph applications scale, performance and cost optimization become critical. This lesson covers strategies to reduce latency, minimize token usage, cache results, batch operations, and tune graph execution.

Onde a Otimização é Importante

Area	Impacto	Otimização
LLM calls	Custo + latency	Caching, batching, model selection
Node count	Latency	Merge trivial nodes, reduce unnecessary nodes
State size	Memory + serialization	Trim state, use efficient data structures
Checkpointing	Latency	Reduce checkpoint frequency
Paralelismo	Throughput	Smart fan-out, avoid contention

Cache de Chamadas LLM

Cache LLM responses for identical inputs:

python

from langchain.globals import set_llm_cache
from langchain.cache import InMemoryCache
import hashlib

# Enable caching
set_llm_cache(InMemoryCache())

# Cache responses based on prompt content
@tool
def cached_search(query: str) -> str:
    """Search with caching for identical queries."""
    return perform_search(query)

# Or use a persistent cache
from langchain.cache import SQLiteCache
set_llm_cache(SQLiteCache(database_path=".llm_cache.db"))

[!NOTA] LLM caching is most effective when users ask similar questions repeatedly. Cache invalidation should be based on time or manual refresh triggers.

Cache Personalizado em Nós

python

import hashlib
import json
from functools import lru_cache

# Simple in-memory cache with TTL
class TTLCache:
    def __init__(self, ttl_seconds: int = 3600):
        self.cache = {}
        self.ttl = ttl_seconds

    def get(self, key: str):
        if key in self.cache:
            value, timestamp = self.cache[key]
            if time.time() - timestamp < self.ttl:
                return value
            del self.cache[key]
        return None

    def set(self, key: str, value):
        self.cache[key] = (value, time.time())

cache = TTLCache(ttl_seconds=300)

def cached_llm_node(state: State) -> dict:
    # Generate cache key from input
    prompt = state["messages"][-1].content
    cache_key = hashlib.md5(prompt.encode()).hexdigest()

    # Check cache
    cached = cache.get(cache_key)
    if cached:
        print("Cache hit!")
        return {"output": cached}

    # Compute and cache
    result = llm.invoke(prompt).content
    cache.set(cache_key, result)
    return {"output": result}

Gerenciamento de Tokens

Orçamento de Tokens

python

import tiktoken

class TokenBudget:
    def __init__(self, model: str, max_tokens: int = 8000):
        self.encoder = tiktoken.encoding_for_model(model)
        self.max_tokens = max_tokens
        self.used_tokens = 0

    def count_tokens(self, text: str) -> int:
        return len(self.encoder.encode(text))

    def can_add(self, text: str) -> bool:
        return self.used_tokens + self.count_tokens(text) <= self.max_tokens

    def add(self, text: str) -> bool:
        tokens = self.count_tokens(text)
        if self.used_tokens + tokens <= self.max_tokens:
            self.used_tokens += tokens
            return True
        return False

def budget_aware_node(state: State) -> dict:
    budget = TokenBudget("gpt-4o", max_tokens=4000)
    trimmed_messages = []

    for msg in reversed(state["messages"]):
        if budget.add(msg.content):
            trimmed_messages.insert(0, msg)

    response = llm.invoke(trimmed_messages)
    return {"response": response.content}

Contagem de Tokens em Streaming

python

class TokenCounter:
    def __init__(self):
        self.prompt_tokens = 0
        self.completion_tokens = 0

    def count_prompt(self, messages: list) -> int:
        enc = tiktoken.encoding_for_model("gpt-4o")
        total = 0
        for msg in messages:
            total += len(enc.encode(msg.content))
        self.prompt_tokens = total
        return total

    def count_completion(self, text: str) -> int:
        enc = tiktoken.encoding_for_model("gpt-4o")
        self.completion_tokens = len(enc.encode(text))
        return self.completion_tokens

[!DICA] Track token usage per-graph-invocation. It helps identify which nodes are most expensive and where to focus optimization efforts.

Agrupamento de Chamadas LLM

When multiple independent LLM calls are needed, batch them:

python

from langchain_core.runnables import RunnableParallel

# Define multiple chains
chain_a = prompt_a | llm | StrOutputParser()
chain_b = prompt_b | llm | StrOutputParser()
chain_c = prompt_c | llm | StrOutputParser()

# Run them in parallel (batched)
combined = RunnableParallel(
    result_a=chain_a,
    result_b=chain_b,
    result_c=chain_c
)

def batched_node(state: State) -> dict:
    results = combined.invoke({
        "topic": state["topic"],
        "question": state["question"]
    })
    return {
        "analysis_a": results["result_a"],
        "analysis_b": results["result_b"],
        "analysis_c": results["result_c"]
    }

Reduzindo Nós Desnecessários

Mesclar Nós de Transformação Triviais

python

# Instead of three nodes for simple transformations:
def clean(state): return {"text": state["text"].strip()}
def lower(state): return {"text": state["text"].lower()}
def truncate(state): return {"text": state["text"][:100]}

# Merge into one:
def process_text(state: State) -> dict:
    return {"text": state["text"].strip().lower()[:100]}

Pular Nós Sem Operação

python

def conditional_node(state: State) -> dict:
    if state.get("skip_processing"):
        return {}  # Return empty — no state changes, minimal overhead
    return {"result": expensive_operation(state)}

Otimização de Checkpointing

python

# Reduce checkpoint frequency
# Instead of checkpointing after every node, batch updates

# Option: Use a single node for sequential operations
def batched_operations(state: State) -> dict:
    # Multiple operations in one node = one checkpoint
    cleaned = state["input"].strip()
    processed = clean(cleaned)
    validated = validate(processed)
    return {"output": validated}

# Option: Disable checkpointing for intermediate nodes
app = builder.compile(
    checkpointer=checkpointer,
    interrupt_before_nodes=[],  # Only checkpoint at specific points
    interrupt_after_nodes=[]
)

Estratégia de Seleção de Modelo

Use the right model for each node:

python

# Expensive model for complex reasoning
reasoning_llm = ChatOpenAI(model="gpt-4o", temperature=0.3)

# Cheap model for simple tasks
fast_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)

# Fastest model for classification/routing
router_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)

def classify_node(state: State) -> dict:
    # Router nodes can use the cheapest model
    category = router_llm.invoke(f"Classify: {state['query']}")
    return {"category": category.content.strip()}

def analyze_node(state: State) -> dict:
    # Complex nodes use the expensive model
    analysis = reasoning_llm.invoke(f"Deep analysis: {state['query']}")
    return {"analysis": analysis.content}

def format_node(state: State) -> dict:
    # Simple formatting uses cheap model
    formatted = fast_llm.invoke(f"Format: {state['analysis']}")
    return {"formatted": formatted.content}

[!SUCESSO] Using different models for different nodes can reduce costs by 5-10x while maintaining quality — use cheap models for routing/simple tasks and expensive models only for complex reasoning.

Lista de Verificação de Otimização

Estratégia	Esforço	Impacto
LLM response caching	Low	High (repeated queries)
Modol selection per node	Low	High (cost)
Token trimming	Medium	High (context management)
Merge trivial nodes	Low	Medium (latency)
Reduce checkpoint frequency	Low	Medium (latency)
Parallel independent work	Medium	High (latency)
Batch multiple LLM calls	Medium	Medium (throughput)
Stream instead of wait	High	High (user experience)

Perguntas Práticas

Practice Question

What is the most impactful optimization for LLM costs?

Practice Question

Which model strategy optimizes cost in a multi-node graph?

Practice Question

What is a good cache TTL for LLM responses?

Practice Question

When should you merge multiple nodes into one?

Practice Question

What is the benefit of streaming LLM tokens?

Practice Question

How does token budgeting help with LLM optimization?

Practice Question

What is a disadvantage of excessive checkpointing?

Practice Question

What does RunnableParallel enable?

Practice Question

What tool can count tokens for OpenAI models?

Practice Question

Which optimizer strategy reduces latency the most?

[!SUCESSO]

Principais Conclusões

Cache LLM responses for repeated queries to reduce cost and latency
Use appropriate model per node: cheap for simple, expensive for complex
Track token usage with tiktoken to manage context windows and budgets
Merge trivial nodes to reduce overhead, but keep meaningful nodes separate
Reduce checkpoint frequency for high-throughput graphs
Parallelize independent work with fan-out for latency improvements
Batch multiple independent LLM calls with RunnableParallel
Streaming improves perceived performance even if total latency is the same

Progresso90%

Execução Paralela

Projeto Final: Agente de Suporte ao Cliente de Produção

Contéudo do Curso

1. Implantação 2. Observabilidade com LangSmith 3. Agentes Hierárquicos 4. RAG Agentic 5. Agentes SQL 6. Padrão de Agente Reflexivo 7. Padrão Planejar e Executar 8. Execução Paralela 9. Otimização 10. Projeto Final: Agente de Suporte ao Cliente de Produção