Cómo construir sistemas de razonamiento agente eficientes podando dinámicamente múltiples rutas de cadena de pensamiento sin perder precisión

En este tutorial, implementamos un marco de poda de cadena de pensamiento agente que genera múltiples rutas de razonamiento en paralelo y las reduce dinámicamente utilizando señales de consenso y detención temprana. Nos centramos en mejorar la eficiencia del razonamiento reduciendo el uso innecesario de tokens y al mismo tiempo preservando la corrección de las respuestas, lo que demuestra que la autoconsistencia y el acuerdo ligero basado en gráficos pueden servir como sustitutos eficaces de la calidad del razonamiento. Diseñamos todo el proceso utilizando un modelo compacto ajustado a instrucciones y muestreo progresivo para simular cómo un agente puede decidir cuándo ha razonado “suficiente”. Consulta los CÓDIGOS COMPLETOS aquí.

!pip -q install -U transformadores aceleran bitsandbytes networkx scikit-learn importar re, tiempo, aleatorio, matemáticas importar numpy como np importar antorcha importar networkx como nx desde transformadores importar AutoTokenizer, AutoModelForCausalLM, GenerationConfig de sklearn.feature_extraction.text importar TfidfVectorizer de sklearn.metrics.pairwise importar cosine_similarity SEED = 7 random.seed(SEED) np.random.seed(SEED) torch.manual_seed(SEED) MODEL_NAME = “Qwen/Qwen2.5-0.5B-Instruct” tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME, use_fast=True) model = AutoModelForCausalLM.from_pretrained( MODEL_NAME, device_map=”auto”, torch_dtype=torch.float16, load_in_4bit=True ) model.eval() SYSTEM = “Eres un solucionador de problemas cuidadoso. Mantenga un razonamiento breve y genere una respuesta numérica final.” FINAL_RE = re.compile(r”Final:\s*([-\d]+(?:\.\d+)?)”)

Configuramos el entorno Colab y cargamos todas las bibliotecas necesarias para un razonamiento agente eficiente. Inicializamos un modelo de lenguaje ligero ajustado a instrucciones con cuantificación para garantizar una ejecución estable en recursos limitados de GPU. También definimos la configuración global, el control de aleatoriedad y el patrón de indicaciones central utilizado a lo largo del tutorial. Consulta los CÓDIGOS COMPLETOS aquí.

def make_prompt(q): return ( f”{SYSTEM}\n\n” f”Problema: {q}\n” f”Razonamiento: (breve)\n” f”Final: ” ) def parse_final_number(texto): m = FINAL_RE.search(texto) if m: return m.group(1).strip() nums = re.findall(r”[-]?\d+(?:\.\d+)?”, texto) devuelve números[-1] if nums else Ninguno def is_correct(pred, gold): si pred es Ninguno: devolver 0 intentar: devolver int(abs(float(pred) – float(gold)) < 1e-9) excepto: return int(str(pred).strip() == str(gold).strip()) def tok_len(texto): devolver len(tokenizer.encode(texto))

Definimos funciones auxiliares que estructuran indicaciones, extraen respuestas numéricas finales y evalúan la corrección en comparación con la verdad básica. Estandarizamos la forma en que se analizan las respuestas para que se puedan comparar diferentes caminos de razonamiento de manera consistente. También presentamos utilidades de conteo de tokens que nos permiten medir posteriormente la eficiencia del razonamiento. Consulta los CÓDIGOS COMPLETOS aquí.

@torch.no_grad() def generate_paths(pregunta, n, max_new_tokens=64, temperatura=0.7, top_p=0.9): solicitud = make_prompt(pregunta) entradas = tokenizer(prompt, return_tensors=”pt”).to(model.device) gen_cfg = GenerationConfig( do_sample=True, temperatura=temperatura, top_p=top_p, max_new_tokens=max_new_tokens, pad_token_id=tokenizer.eos_token_id, eos_token_id=tokenizer.eos_token_id, num_return_sequences=n ) out = model.generate(**inputs, generate_config=gen_cfg) Prompt_tok = entradas[“input_ids”].forma[1]

caminos = []
para i dentro del rango (fuera de forma)[0]): secuencia = fuera[i]
gen_ids = secuencia[prompt_tok:]
finalización = tokenizer.decode(gen_ids, skip_special_tokens=True) paths.append({ “prompt_tokens”: int(prompt_tok), “gen_tokens”: int(gen_ids.shape[0]), “finalización”: finalización }) rutas de retorno

Implementamos una generación rápida de múltiples muestras que produce varias rutas de razonamiento en una sola llamada de modelo. Extraemos solo la continuación generada para aislar el resultado del razonamiento para cada ruta. Almacenamos el uso y las finalizaciones de tokens en un formato estructurado para respaldar las decisiones de poda posteriores. Consulta los CÓDIGOS COMPLETOS aquí.

def consenso_strength(finalizaciones, sim_threshold=0.22): si len(finalizaciones) <= 1: regresar [0.0] * len(compleciones) vec = TfidfVectorizer(ngram_range=(1,2), max_features=2500) X = vec.fit_transform(compleciones) S = cosine_similarity(X) G = nx.Graph() n = len(compleciones) G.add_nodes_from(range(n)) para i en range(n): para j en range(i+1, n): w = flotador (S[i, j]) si w >= sim_threshold: G.add_edge(i, j, peso=w) fuerza = [0.0] * n para u, v, d en G.edges(data=True): w = float(d.get(“weight”, 0.0)) fuerza[u] += w fuerza[v] += w fuerza de retorno

Construimos un mecanismo de consenso ligero utilizando un gráfico de similitud sobre rutas de razonamiento generadas. Calculamos puntuaciones de similitud por pares y las convertimos en una señal de intensidad basada en gráficos para cada ruta. Nos permite aproximarnos al acuerdo entre trayectorias de razonamiento sin costosas llamadas a modelos. Consulta los CÓDIGOS COMPLETOS aquí.

def pick_final_answer(rutas): respuestas = [parse_final_number(p[“completion”]) para p en caminos]fortalezas = consenso_strength([p[“completion”] para p en rutas]) grupos = {} para i, a en enumerar(respuestas): si a es Ninguno: continuar groups.setdefault(a, {“idx”: []”fuerza”: 0.0, “tokens”: 0}) grupos[a][“idx”].append(i) grupos[a][“strength”] += fortalezas[i]
grupos[a][“tokens”] += caminos[i][“gen_tokens”]

si no es grupos: devuelve Ninguno, {“respuestas”: respuestas, “fortalezas”: fortalezas} clasificado = ordenado( grupos.items(), clave=lambda kv: (len(kv[1][“idx”]), kilovoltios[1][“strength”]-kv[1][“tokens”]), inversa=Verdadero ) mejor_respuesta = clasificada[0][0]

best_indices = clasificado[0][1][“idx”]

best_i = ordenado(best_indices, key=lambda i: (rutas[i][“gen_tokens”]-fortalezas[i]))[0]

return best_answer, {“respuestas”: respuestas, “fortalezas”: fortalezas, “best_i”: best_i} def pruned_agent_answer (pregunta, tamaño de lote = 2, k_max = 10, max_new_tokens = 64, temperatura = 0,7, top_p = 0,9, stop_min_samples = 4, stop_ratio = 0,67, stop_margin = 2): rutas = []
Prompt_tokens_once = tok_len(make_prompt(pregunta)) total_gen_tokens = 0 mientras len(rutas) < k_max: n = min(tamaño_lote, k_max - len(rutas)) new_paths = generar_rutas( pregunta, n=n, max_new_tokens=max_new_tokens, temperatura=temperatura, top_p=top_p ) rutas.extend(nuevas_rutas) total_gen_tokens += suma(p["gen_tokens"] para p en new_paths) si len(paths) >= stop_min_samples: respuestas = [parse_final_number(p[“completion”]) para p en rutas]cuenta = {} para a en respuestas: si a es Ninguno: continúa contando[a] = counts.get(a, 0) + 1 si cuenta: sorted_counts = sorted(counts.items(), key=lambda kv: kv[1]reverso=Verdadero) top_a, top_c = recuentos_clasificados[0]
segundo_c = recuentos_ordenados[1][1] if len(sorted_counts) > 1 else 0 if top_c >= math.ceil(stop_ratio * len(paths)) y (top_c – second_c) >= stop_margin: final, dbg = pick_final_answer(paths) return { “final”: final, “paths”: caminos, “early_stopped_at”: len(paths), “tokens_total”: int(prompt_tokens_once * len(rutas) + total_gen_tokens), “debug”: dbg } final, dbg = pick_final_answer(rutas) return { “final”: final, “paths”: rutas, “early_stopped_at”: Ninguno, “tokens_total”: int(prompt_tokens_once * len(rutas) + total_gen_tokens), “debug”: dbg}

Implementamos la lógica central de poda agente que agrupa las rutas de razonamiento según las respuestas finales y las clasifica utilizando señales de consenso y eficiencia. Introducimos un muestreo progresivo con detención temprana para terminar la generación una vez que surja suficiente confianza. Luego seleccionamos una respuesta final que equilibre la solidez del acuerdo y el uso mínimo de tokens. Consulta los CÓDIGOS COMPLETOS aquí.

def baseline_answer(pregunta, k=10, max_new_tokens=64): rutas = generar_rutas(pregunta, n=k, max_new_tokens=max_new_tokens) Prompt_tokens_once = tok_len(make_prompt(pregunta)) total_gen_tokens = suma(p[“gen_tokens”] para p en caminos) respuestas = [parse_final_number(p[“completion”]) para p en rutas]cuenta = {} para a en respuestas: si a es Ninguno: continúa contando[a] = counts.get(a, 0) + 1 final = max(counts.items(), clave=lambda kv: kv[1])[0] si cuenta de lo contrario Ninguno return { “final”: final, “paths”: rutas, “tokens_total”: int(prompt_tokens_once * k + total_gen_tokens) } DATOS = [
{“q”: “If a store sells 3 notebooks for $12, how much does 1 notebook cost?”, “a”: “4”},
{“q”: “What is 17*6?”, “a”: “102”},
{“q”: “A rectangle has length 9 and width 4. What is its area?”, “a”: “36”},
{“q”: “If you buy 5 apples at $2 each, how much do you pay?”, “a”: “10”},
{“q”: “What is 144 divided by 12?”, “a”: “12”},
{“q”: “If x=8, what is 3x+5?”, “a”: “29”},
{“q”: “A jar has 30 candies. You eat 7. How many remain?”, “a”: “23”},
{“q”: “If a train travels 60 km in 1.5 hours, what is its average speed (km/h)?”, “a”: “40”},
{“q”: “Compute: (25 – 9) * 3”, “a”: “48”},
{“q”: “What is the next number in the pattern: 2, 4, 8, 16, ?”, “a”: “32”},
]

base_acc, base_tok = [], []
prun_acc, prun_tok = [], []

para el elemento en DATOS: b = baseline_answer(elemento[“q”]k=8, max_new_tokens=56) base_acc.append(is_correct(b[“final”]artículo[“a”])) base_tok.append(b[“tokens_total”]) p = respuesta_agente_podada(elemento[“q”]max_new_tokens=56) prun_acc.append(is_correct(p[“final”]artículo[“a”])) prun_tok.append(p[“tokens_total”]) print(“Precisión de línea base:”, float(np.mean(base_acc))) print(“Tokens promedio de línea base:”, float(np.mean(base_tok))) print(“Precisión podada:”, float(np.mean(prun_acc))) print(“Tokens promedio podados:”, float(np.mean(prun_tok)))

Comparamos el enfoque agencial podado con una línea de base fija de autoconsistencia. Evaluamos ambos métodos en cuanto a precisión y consumo de tokens para cuantificar las ganancias de eficiencia de la poda. Concluimos informando métricas agregadas que demuestran cómo la poda dinámica preserva la corrección al tiempo que reduce el costo de razonamiento.

En conclusión, demostramos que la poda agente puede reducir significativamente el consumo efectivo de tokens sin sacrificar la precisión al detener el razonamiento una vez que surge un consenso suficiente. Demostramos que la combinación de autoconsistencia, gráficos de consenso basados ​​en similitudes y heurísticas de parada temprana proporciona un enfoque práctico y escalable para la eficiencia del razonamiento en sistemas agentes. Este marco sirve como base para comportamientos de agencia más avanzados, como la poda de generación media, el razonamiento consciente del presupuesto y el control adaptativo sobre la profundidad del razonamiento en agentes de IA del mundo real.

Consulta los CÓDIGOS COMPLETOS aquí. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 100.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.