O que é Replanejamento
Planejar é fácil. O plano raramente sobrevive ao contato com a realidade. Replanejamento automático é a capacidade do sistema de detectar que o plano original não está mais válido e gerar um novo plano — mantendo o objetivo final em vista.
Retry vs Replanning — Qual a Diferença?
Retry
- Mesma abordagem, tenta de novo
- Assume que a falha foi transitória
- Mantém o plano intacto
- Serve para: timeout, rate limit, erro 5xx
Replanning
- Abordagem diferente, objetivo igual
- Reconhece que o plano falhou estruturalmente
- Gera novo plano com nova estratégia
- Serve para: premissa inválida, recurso indisponível, resultado divergente
Quando o Plano Fica Inválido
Premissa não mais verdadeira
Agente assumiu que documento X estava disponível, mas foi deletado durante a execução.
Recurso indisponível
API externa que era central ao plano está offline. Precisa de caminho alternativo.
Resultado intermediário divergiu
Step 3 retornou dados que invalidam steps 4 e 5. Precisa de replanejamento a partir do step 4.
Objetivo mudou
Novo input do usuário redirecionou o objetivo enquanto o agente estava executando o plano antigo.
Detecção de Desvio
Antes de replanejar, é preciso detectar que o plano está desviando. Isso exige monitoramento contínuo durante a execução — comparar o estado atual com o estado esperado a cada step.
Python — Monitor de Desvio
from dataclasses import dataclass
from typing import Optional
from enum import Enum
class DeviationLevel(Enum):
NONE = "none"
MINOR = "minor" # Continua, ajuste leve
MODERATE = "moderate" # Replanning parcial
CRITICAL = "critical" # Replanning total ou escalação
@dataclass
class DeviationReport:
level: DeviationLevel
step_id: str
expected: dict
actual: dict
deviation_score: float # 0.0 a 1.0
description: str
class DeviationDetector:
def __init__(self, threshold_minor: float = 0.2,
threshold_moderate: float = 0.5,
threshold_critical: float = 0.8):
self.threshold_minor = threshold_minor
self.threshold_moderate = threshold_moderate
self.threshold_critical = threshold_critical
async def check(
self,
step_id: str,
expected_state: dict,
actual_state: dict
) -> DeviationReport:
"""Compara estado esperado vs atual e classifica desvio."""
score = await self._compute_deviation_score(
expected_state,
actual_state
)
if score < self.threshold_minor:
level = DeviationLevel.NONE
elif score < self.threshold_moderate:
level = DeviationLevel.MINOR
elif score < self.threshold_critical:
level = DeviationLevel.MODERATE
else:
level = DeviationLevel.CRITICAL
return DeviationReport(
level=level,
step_id=step_id,
expected=expected_state,
actual=actual_state,
deviation_score=score,
description=self._describe_deviation(expected_state, actual_state)
)
async def _compute_deviation_score(
self,
expected: dict,
actual: dict
) -> float:
"""
Usa LLM para comparar semântica dos estados.
Score 0.0 = idênticos, 1.0 = completamente divergentes.
"""
prompt = f"""Compare os estados e retorne um score de desvio de 0.0 a 1.0.
0.0 = resultados equivalentes, 1.0 = completamente divergentes.
Estado esperado: {expected}
Estado atual: {actual}
Responda APENAS com um número decimal entre 0.0 e 1.0."""
response = await llm.complete(prompt)
return float(response.strip())
def _describe_deviation(self, expected: dict, actual: dict) -> str:
missing = set(expected.keys()) - set(actual.keys())
extra = set(actual.keys()) - set(expected.keys())
parts = []
if missing:
parts.append(f"Campos ausentes: {missing}")
if extra:
parts.append(f"Campos inesperados: {extra}")
return "; ".join(parts) or "Desvio semântico"
Score 0.0–0.2
Nenhum desvio — continua
Score 0.2–0.5
Desvio leve — ajuste fino
Score 0.5–1.0
Desvio crítico — replanning
Triggers de Replanning
Um trigger é uma condição objetiva que, quando satisfeita, dispara o processo de replanejamento. Triggers bem definidos evitam replanejamentos desnecessários (muito frequentes) ou tardios (que deixam o sistema em desvio por muito tempo).
Desvio de Score Acima do Threshold
O monitor de desvio retornou score > 0.5. O estado atual divergiu o suficiente do esperado para justificar replanejamento.
Step Bloqueado N Vezes
Um step específico falhou 3+ vezes com retry. Não é falha transitória — a abordagem está errada e precisa de nova estratégia para aquele step.
Dependência Indisponível
Um recurso ou API essencial ao plano original está offline ou inacessível. O plano depende desse recurso e precisa ser reescrito sem ele.
Novo Input do Usuário
O usuário adicionou contexto ou mudou a prioridade enquanto o agente executava. O plano precisa incorporar a nova informação.
Python — Sistema de Triggers
class ReplanningTriggerSystem:
def __init__(self):
self.triggers = []
def register(self, trigger_fn, name: str = "unnamed"):
self.triggers.append((name, trigger_fn))
async def evaluate(self, context: dict) -> Optional[str]:
"""Avalia todos os triggers. Retorna nome do primeiro ativado."""
for name, trigger_fn in self.triggers:
if await trigger_fn(context):
return name
return None
# Configuração
trigger_system = ReplanningTriggerSystem()
trigger_system.register(
lambda ctx: ctx["deviation_score"] > 0.5,
"high_deviation"
)
trigger_system.register(
lambda ctx: ctx["step_failures"].get(ctx["current_step"], 0) >= 3,
"step_blocked"
)
trigger_system.register(
lambda ctx: not ctx["dependencies_available"],
"dependency_down"
)
trigger_system.register(
lambda ctx: ctx.get("new_user_input") is not None,
"user_redirect"
)
# No loop de execução
triggered = await trigger_system.evaluate(execution_context)
if triggered:
print(f"Replanning disparado por: {triggered}")
new_plan = await replanner.generate(execution_context)
Estratégias de Replanning
Nem todo replanejamento precisa jogar fora tudo. A estratégia certa depende da amplitude do desvio: replanning total, parcial ou escalada para humano.
Replanning Parcial
Apenas os steps afetados pelo desvio são replanejados. Steps concluídos com sucesso são mantidos.
Quando usar: desvio localizado em 1-2 steps
Replanning Total
Descarta o plano atual e gera um novo do zero, com base no objetivo original e no estado atual do mundo.
Quando usar: premissa central inválida
Escalação
O replanejamento automático não consegue resolver. Suspende e passa para humano com contexto completo.
Quando usar: desvio crítico ou irreversível
Python — Replanner Adaptativo
class AdaptiveReplanner:
def __init__(self, llm, escalation_manager):
self.llm = llm
self.escalation = escalation_manager
async def replan(
self,
original_goal: str,
original_plan: list,
completed_steps: list,
failed_step: str,
current_state: dict,
trigger_reason: str
) -> list:
"""
Decide estratégia e gera novo plano.
Returns: lista de steps do novo plano.
"""
# Quantos steps foram concluídos com sucesso?
completion_ratio = len(completed_steps) / len(original_plan)
# Determina estratégia
if completion_ratio > 0.7:
# Mais de 70% concluído — replanning parcial
strategy = "partial"
context_prefix = f"Steps já concluídos (manter): {completed_steps}"
elif completion_ratio < 0.3 or trigger_reason == "dependency_down":
# Pouco progresso ou dependência core fora — replanning total
strategy = "total"
context_prefix = "Replanejar completamente desde o início"
else:
strategy = "partial"
context_prefix = f"Steps já concluídos (manter): {completed_steps}"
# Gera novo plano via LLM
prompt = f"""
Objetivo: {original_goal}
{context_prefix}
Estado atual: {current_state}
Motivo do replanejamento: {trigger_reason}
Step que falhou: {failed_step}
Gere um novo plano de execução (lista de steps JSON) para alcançar o objetivo
dado o estado atual. Seja conciso e prático.
Formato: [{{"step_id": "...", "action": "...", "agent": "..."}}]
"""
response = await self.llm.complete(prompt)
try:
new_steps = json.loads(response)
return new_steps
except json.JSONDecodeError:
# LLM não retornou JSON válido — escalar
await self.escalation.request_approval(
EscalationCheckpoint(
task_id="replanning_failed",
reason="Replanner não conseguiu gerar plano válido",
context={"state": current_state, "trigger": trigger_reason},
action_proposed="Revisão manual do plano de execução",
risk_level="high"
)
)
raise ReplanningError("Escalação para humano após falha de replanning")
Limite de Replanning
Replanejamento ilimitado é um anti-pattern perigoso. Um agente que continua replanejando indefinidamente pode consumir recursos ilimitados sem nunca convergir. É necessário definir limites explícitos e o que acontece quando eles são atingidos.
Python — Replanning com Limite
@dataclass
class ReplanningBudget:
max_replannings: int = 3 # Máximo de replanejamentos totais
max_consecutive: int = 2 # Máximo consecutivos sem progresso
max_cost_increase: float = 2.0 # Custo máximo = 2x do original
max_time_increase: float = 3.0 # Tempo máximo = 3x do original
class ResilientOrchestrator:
def __init__(self, budget: ReplanningBudget):
self.budget = budget
self.replanning_count = 0
self.consecutive_replannings = 0
self.original_estimated_cost = 0
self.original_estimated_time = 0
async def execute(self, goal: str, initial_plan: list) -> dict:
plan = initial_plan
self.original_estimated_cost = estimate_cost(plan)
self.original_estimated_time = estimate_time(plan)
last_completed_count = 0
completed_steps = []
while plan:
step = plan[0]
try:
result = await execute_step(step)
completed_steps.append(step)
plan = plan[1:] # Remove step concluído
self.consecutive_replannings = 0 # Reset após sucesso
except Exception as e:
# Verifica se pode replanejar
if not self._can_replan(completed_steps, last_completed_count):
raise BudgetExceededError(
f"Budget de replanning esgotado após "
f"{self.replanning_count} replanejamentos. "
f"Escalando para humano."
)
# Replanejar
trigger = classify_failure(e)
plan = await replanner.replan(
goal, plan, completed_steps, step["step_id"],
get_current_state(), trigger
)
self.replanning_count += 1
self.consecutive_replannings += 1
last_completed_count = len(completed_steps)
return {"completed_steps": completed_steps, "replannings": self.replanning_count}
def _can_replan(self, completed_steps: list, last_count: int) -> bool:
# Limite total
if self.replanning_count >= self.budget.max_replannings:
print(f"Limite total de replanejamentos atingido: {self.budget.max_replannings}")
return False
# Limite consecutivo (sem progresso)
if self.consecutive_replannings >= self.budget.max_consecutive:
print(f"Replanejamentos consecutivos sem progresso: {self.budget.max_consecutive}")
return False
# Limite de custo
current_cost = get_current_cost()
if current_cost > self.original_estimated_cost * self.budget.max_cost_increase:
print(f"Custo {current_cost:.2f} excede limite de {self.budget.max_cost_increase}x")
return False
return True
Valores Recomendados para Produção
3
Max replanejamentos
2
Max consecutivos
2x
Limite de custo
3x
Limite de tempo
Exemplos Reais
Como o replanejamento automático aparece em sistemas de produção — com contexto real, triggers específicos e estratégias aplicadas.
Agente de Pesquisa de Mercado
Plano: buscar dados de 5 fontes → consolidar → gerar relatório
Trigger
API do Bloomberg está offline (503 após 3 retries)
Estratégia
Replanning parcial: substitui Bloomberg por Reuters + Yahoo Finance
Resultado
Relatório entregue com nota de fonte alternativa. Sem intervenção humana.
Agente de Geração de Código
Plano: escrever módulo → testar → fazer PR
Trigger
Testes falharam 2x consecutivas. Código gerado tem bug lógico.
Estratégia
Replanning total do módulo: abordagem diferente + análise de erro dos testes como contexto
Resultado
Nova abordagem com TDD reverso: escreve tests first, depois implementa para passar.
Pipeline de Onboarding de Cliente
Plano: verificar documentos → validar KYC → ativar conta
Trigger
KYC retornou "identidade não verificável" — premissa de que cliente tem documentos válidos falhou
Estratégia
Escalação imediata: ação irreversível (ativação de conta) não pode prosseguir sem certeza
Resultado
Ticket criado para compliance humano. Conta mantida em pending. Log completo preservado.
Resumo do Módulo
Conceitos centrais
- → Retry tenta de novo; replanning muda a abordagem
- → Detecção de desvio via score semântico LLM
- → Triggers objetivos evitam replanejamento excessivo
- → Estratégias: parcial, total, escalação
Na prática
- → Budget explícito: 3 replanejamentos, 2x custo máximo
- → Replanning consecutivo sem progresso → escalação
- → Manter steps bem-sucedidos no replanning parcial
- → Ações irreversíveis sempre escalam para humano