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.

À medida que as aplicações LangGraph crescem, a otimização de desempenho e de custo torna-se crítica. Esta lição cobre estratégias para reduzir latência, minimizar o consumo de tokens, armazenar resultados em cache, agrupar operações e ajustar a execução do grafo.

Onde a Otimização é Importante

Área	Impacto	Otimização
Chamadas de LLM	Custo + latência	Cache, agrupamento e seleção de modelo
Número de nós	Latência	Mescle nós triviais, reduza nós desnecessários
Tamanho do estado	Memória + serialização	Reduza o estado, use estruturas de dados eficientes
Checkpointing	Latência	Reduza a frequência de checkpointing
Paralelismo	Throughput	Fan-out inteligente, evite contenção

Cache de Chamadas LLM

Armazene em cache as respostas de LLM para entradas idênticas:

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

O cache de LLM é mais eficaz quando os usuários fazem perguntas semelhantes repetidamente. A invalidação de cache deve ser baseada em tempo ou em gatilhos de atualização manual.

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

Acompanhe o consumo de tokens por invocação de grafo. Isso ajuda a identificar quais nós são mais caros e onde concentrar os esforços de otimização.

Agrupamento de Chamadas LLM

Quando forem necessárias múltiplas chamadas independentes de LLM, agrupe-as:

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 o modelo certo para cada nó:

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

Usar modelos diferentes para nós diferentes pode reduzir os custos em 5-10x mantendo a qualidade: use modelos baratos para roteamento/tarefas simples e modelos caros apenas para raciocínio complexo.

Lista de Verificação de Otimização

Estratégia	Esforço	Impacto
Cache de respostas de LLM	Baixo	Alto (consultas repetidas)
Seleção de modelo por nó	Baixo	Alto (custo)
Truncamento de tokens	Médio	Alto (gerenciamento de contexto)
Mesclar nós triviais	Baixo	Médio (latência)
Reduzir frequência de checkpointing	Baixo	Médio (latência)
Trabalho independente em paralelo	Médio	Alto (latência)
Agrupar múltiplas chamadas de LLM	Médio	Médio (throughput)
Streaming em vez de espera	Alto	Alto (experiência do usuário)