Descomponer tareas grandes en pase local tipado y pase de integracion sobre resumenes, con schemas de transicion entre pases.
Diseñar e implementar el procesamiento de una tarea grande como una cadena de pases en lugar de un único mega-prompt. El patrón canónico tiene dos etapas: un pase local tipado que procesa cada unidad (archivo, ticket, registro) de forma aislada y emite un resumen estructurado contra un schema, y un pase de integración que solo consume esos resúmenes —nunca los datos crudos— para producir el resultado final. Entre pases existe un schema de transición explícito que define qué viaja, en qué forma, y cómo se representa el error por unidad. La capacidad sustituye saturación de atención y costo cuadrático por paralelización tipada y trazable. [DOC]
Inputs requeridos (sin ellos: {VACIO_CRITICO}, parar y pedir — ver eval empty_input): [DOC]
Outputs producidos: [DOC]
ok/error por unidad).single_pass_fits). [INFERENCIA]false_positive_unrelated). [DOC]raws_in_integration_rejected, global_exception_rejected). [DOC]ok / error) con detalle del fallo. Sin schema no hay cadena, hay pegamento.status="error") para que un fallo aislado no tumbe el lote ni contamine la síntesis; cuesta disciplina de tipado en cada pase local. [INFERENCIA]# GOOD: pase local tipado por unidad + integración solo sobre resúmenes.
from pydantic import BaseModel
from typing import Literal
class UnitSummary(BaseModel):
unit_id: str
status: Literal["ok", "error"]
findings: list[str] = []
error_detail: str | None = None
def local_pass(unit: SourceFile) -> UnitSummary:
# Ve UNA sola unidad. El fallo se tipa, no se lanza.
try:
findings = analyze(unit.content)
return UnitSummary(unit_id=unit.id, status="ok", findings=findings)
except AnalysisError as exc:
return UnitSummary(unit_id=unit.id, status="error", error_detail=str(exc))
# Schema de transición: colección tipada de resúmenes (nunca crudos).
summaries: list[UnitSummary] = [local_pass(u) for u in units] # paralelizable
def integration_pass(summaries: list[UnitSummary]) -> Report:
ok = [s for s in summaries if s.status == "ok"]
failed = [s for s in summaries if s.status == "error"]
# El pase 2 razona SOLO sobre resúmenes tipados.
return synthesize(ok, failures=failed)# ANTI: mega-prompt que concatena todos los crudos en una sola pasada.
# Satura la atención, no paraleliza, y un fallo en un archivo contamina todo.
blob = "\n\n".join(read(f) for f in fifty_files) # 50 archivos crudos juntos
result = model(f"Analiza todo esto y dame el reporte:\n{blob}")
# Sin schema por pase, sin estado de error por unidad, sin transición tipada.Cualquier "no" es bloqueante: corrige el diseño antes de entregar, no lo marques completo. [DOC]
status="error", error_detail) y continúa; el pase 2 reporta los fallos sin abortar (eval error_isolation). [DOC]upgrade_safety_case). [DOC]{VACIO_CRITICO}: para y pide, no auto-completes el lote (eval empty_input). [DOC]assets/ define el contrato determinístico para justificación vs single-pass, unidad atómica, schema local, schema de transición, integración sobre resúmenes y errores tipados. [CÓDIGO]scripts/check.sh valida fixtures locales sin red, tiempo real ni aleatoriedad. [CÓDIGO]scripts/validate_prompt_chaining_design.py rechaza diseños donde el pase de integración consume crudos, falta schema, el pase local procesa varias unidades, no hay error tipado o Guardian aprueba un diseño bloqueado. [CÓDIGO]katas-multipass-prompt-chainingworkflow-forgeoutput-engineeringfdad39c
If you maintain this skill, you can claim it as your own. Once claimed, you can manage eval scenarios, bundle related skills, attach documentation or rules, and ensure cross-agent compatibility.