En este tutorial, exploramos el conjunto de datos lambda/hermes-agent-reasoning-traces para comprender cómo piensan los modelos basados en agentes, utilizan herramientas y generan respuestas en conversaciones de varios turnos. Comenzamos cargando e inspeccionando el conjunto de datos, examinando su estructura, categorías y formato conversacional para tener una idea clara de la información disponible. Luego construimos analizadores simples para extraer componentes clave, como rastros de razonamiento, llamadas a herramientas y respuestas de herramientas, lo que nos permite separar el pensamiento interno de las acciones externas. Además, analizamos patrones como la frecuencia de uso de herramientas, la duración de las conversaciones y las tasas de error para comprender mejor el comportamiento de los agentes. También creamos visualizaciones para resaltar estas tendencias y hacer el análisis más intuitivo. Finalmente, preparamos el conjunto de datos para el entrenamiento convirtiéndolo a un formato compatible con modelos, haciéndolo adecuado para tareas como el ajuste fino supervisado.
!pip -q install -U conjuntos de datos pandas matplotlib transformadores marinos acelerar trl importar json, re, aleatorio, ajuste de texto de colecciones importar Contador, defaultdict importar pandas como pd importar numpy como np importar matplotlib.pyplot como plt desde conjuntos de datos importar load_dataset, concatenate_datasets random.seed(0) CONFIG = "kimi" ds = load_dataset("lambda/hermes-agent-reasoning-traces", CONFIG, split="train") print(ds) print("Config:", CONFIG, "| Fields:", ds.column_names) print("Categorías:", sorted(set(ds["category"]))) COMPARE_BOTH = False si COMPARE_BOTH: ds_kimi = load_dataset("lambda/hermes-agent-reasoning-traces", "kimi", split="train") ds_glm = load_dataset("lambda/hermes-agent-reasoning-traces", "glm-5.1", split="train") ds_kimi = ds_kimi.add_column("source", ["kimi"] * len(ds_kimi)) ds_glm = ds_glm.add_column("fuente", ["glm-5.1"] * len(ds_glm)) ds = concatenate_datasets([ds_kimi, ds_glm]).shuffle(seed=0) print("Combinado:", ds, "→ recuentos:", Counter(ds["fuente"])) muestra = ds[0]print("n=== Muestra 0 ===") print("id:", muestra["id"]) print("categoría:", muestra["categoría"], "/", muestra["subcategoría"]) print("tarea:", muestra["tarea"]) print("turnos:", len(muestra["conversaciones"])) print("sistema[0]:", muestra["conversaciones"][0]["valor"][:220], "…n")
Instalamos todas las bibliotecas necesarias e importamos los módulos necesarios para configurar nuestro entorno. Luego cargamos el conjunto de datos lambda/hermes-agent-reasoning-traces e inspeccionamos su estructura, campos y categorías. Opcionalmente, también combinamos múltiples configuraciones de conjuntos de datos y examinamos una muestra para comprender el formato conversacional.
THINK_RE = re.compile(r"(.*?)", re.DOTALL) TOOL_CALL_RE = re.compile(r"s*({.*?})s*", re.DOTALL) TOOL_RESP_RE = re.compile(r"s*(.*?)s*", re.DOTALL) def parse_assistant(valor: str) -> dict: pensamientos = [t.strip() para t en THINK_RE.findall(valor)] llamadas =[]para raw en TOOL_CALL_RE.findall(value): intente: call.append(json.loads(raw)) excepto json.JSONDecodeError: call.append({"name": "", "arguments": {}}) final = TOOL_CALL_RE.sub("", THINK_RE.sub("", value)).strip() return {"thinkts": pensamientos, "tool_calls": llamadas, "final": final} def parse_tool(valor: str): raw = TOOL_RESP_RE.search(valor) si no es raw: devolver {"raw": valor} cuerpo = raw.group(1) intentar: devolver json.loads(cuerpo) excepto: devolver {"raw": cuerpo} first_gpt = next(t for t in sample["conversations"] if t["from"] == "gpt") p = parse_assistant(first_gpt["value"]) print("Vista previa del pensamiento:", (p["pensamientos"][0][:160] + "…") if p["pensamientos"] else "(ninguno)") print("Llamadas a herramientas :", [(c.get("nombre"), list(c.get("argumentos", {}).keys())) para c en p["tool_calls"]])
Definimos analizadores basados en expresiones regulares para extraer rastros de razonamiento, llamadas a herramientas y respuestas de herramientas del conjunto de datos. Procesamos los mensajes del asistente para separar pensamientos, acciones y resultados finales de forma estructurada. Luego probamos el analizador en una conversación de muestra para verificar que la extracción funcione correctamente.
N = 3000 sub = ds.select(range(min(N, len(ds)))) llamadas_herramientas = Contador() anchos_paralelos = Contador() pensamientos_por_turno =[]llamadas_per_traj =[]errores_per_traj =[]vueltas_per_traj =[]cat_counts = Counter() para ex in sub: cat_counts[ex["categoría"]] += 1 n_calls = n_err = 0 Turns_per_traj.append(len(ex["conversaciones"])) para t en ex["conversaciones"]: if t["from"] == "gpt": p = parse_assistant(t["value"]) pensamientos_por_turno.append(len(p["pensamientos"])) if p["tool_calls"]: anchos_paralelos[len(p["tool_calls"])] += 1 for c in p["tool_calls"]: tool_calls[c.get("name", "")] += 1 n_calls += len(p["tool_calls"]) elif t["from"] == "tool": r = parse_tool(t["value"]) blob = json.dumps(r).lower() si "error" en blob o '"exit_code": 1' en blob o "rastreo" en blob: n_err += 1 call_per_traj.append(n_calls) errores_per_traj.append(n_err) print(f"nScanned {len(sub)} trayectorias") print(f"Promedio de giros/traj: {np.mean(turns_per_traj):.1f}") print(f"Promedio de llamadas de herramienta/traj: {np.mean(calls_per_traj):.1f}") print(f"% con >=1 error: {100*np.mean([e>0 for e in errores_per_traj]):.1f}%") print(f"% giros paralelos: {100*sum(v para k,v en paralelo_widths.items() si k>1)/max(1,sum(parallel_widths.values())):.1f}%") print("Las 10 herramientas principales:", tool_calls.most_common(10)) fig, axes = plt.subplots(2, 2, figsize=(13, 9)) top = tool_calls.most_common(15) axes[0,0].barh([t for t,_ in top][::-1], [c for _,c in top][::-1], color="teal") axes[0,0].set_title("Las 15 mejores herramientas por volumen de llamadas") axes[0,0].set_xlabel("calls") ks = sorted(parallel_widths) axes[0,1].bar([str(k) para k en ks], [parallel_widths[k] para k en ks], color="coral") axes[0,1].set_title("Llamadas a herramientas por turno de asistente (ancho paralelo)") axes[0,1].set_xlabel("# llamadas a herramientas en un turno"); axes[0,1].set_ylabel("count") axes[0,1].set_yscale("log") axes[1,0].hist(turns_per_traj, bins=40, color="steelblue") axes[1,0].set_title("Duración de la conversación"); ejes[1,0].set_xlabel("gira") gatos, vals = zip(*cat_counts.most_common()) ejes[1,1].pie(vals, etiquetas=gatos, autopct="%1.0f%%", startangle=90) ejes[1,1].set_title("Distribución de categorías") plt.tight_layout(); plt.mostrar()
Realizamos análisis de todo el conjunto de datos para medir el uso de herramientas, la duración de las conversaciones y los patrones de error. Agregamos estadísticas de varias muestras para comprender el comportamiento general de los agentes. También creamos visualizaciones para resaltar tendencias como la frecuencia de las herramientas, las llamadas paralelas y la distribución de categorías.
def render_trace(ex, max_chars=350): print(f"n{'='*72}nTASK [{ex['category']} / {ex['subcategory']}]: {ex['task']}n{'='*72}") for t in ex["conversations"]: rol = t["from"] if rol == "sistema": continuar si rol == "humano": print(f"n[USUARIO]n{textwrap.shorten(t['valor'], 600)}") elif role == "gpt": p = parse_assistant(t["valor"]) para th en p["pensamientos"]: print(f"n[PENSAR]n{textwrap.shorten(th, max_chars)}") para c en p["tool_calls"]: args = json.dumps(c.get("argumentos", {}))[:200] print(f"[LLAMADA] {c.get('nombre')}({args})") if p["final"]: print(f"n[RESPUESTA]n{textwrap.shorten(p['final'], max_chars)}") elif role == "herramienta": print(f"[TOOL_RESPONSE] {textwrap.shorten(t['value'], 220)}") print("="*72) idx = int(np.argmin(np.abs(np.array(turns_per_traj) – 10))) render_trace(sub[idx]) def get_tool_schemas(ex): try: return json.loads(ex["tools"]) excepto: return[]esquemas = get_tool_schemas(muestra) print(f"nLa muestra 0 tiene {len(schemas)} herramientas disponibles") para s en esquemas[:3]: fn = s.get("función", {}) print(" -", fn.get("nombre"), "—", (fn.get("descripción") o "")[:80]) ROLE_MAP = {"system": "system", "human": "user", "gpt": "assistant", "tool": "tool"} def to_openai_messages(conv): return [{"role": ROLE_MAP[t["from"]], "content": t["value"]} for t in conv] example_msgs = to_openai_messages(sample["conversations"]) print("nPrimeros 2 mensajes de OpenAI:") para m in example_msgs[:2]: print(" ", m["rol"], "→", m["content"][:120].replace("n", " "), "…")
Creamos utilidades para representar seguimientos completos de conversaciones en un formato legible para una inspección más profunda. También extraemos esquemas de herramientas y convertimos el conjunto de datos a un formato de mensaje estilo OpenAI para que sea compatible con los canales de capacitación. Esto nos ayuda a comprender mejor tanto la estructura de las herramientas como cómo se pueden estandarizar las conversaciones.
de transformadores importar AutoTokenizer TOK_ID = "Qwen/Qwen2.5-0.5B-Instruct" tok = AutoTokenizer.from_pretrained(TOK_ID) def build_masked(conv, tokenizer, max_len=2048): msgs = to_openai_messages(conv) for m en mensajes: if m["role"] == "tool": m["role"] = "usuario" m["content"] = "[SALIDA DE HERRAMIENTA]n" + m["content"] input_ids, etiquetas =[],[]para m en mensajes: texto = tokenizer.apply_chat_template([m], tokenize=False, add_generación_prompt=False) ids = tokenizer.encode(text, add_special_tokens=False) input_ids.extend(ids) etiquetas.extend(ids if m["role"] == "asistente" else [-100] * len(ids)) return input_ids[:max_len], etiquetas[:max_len] ids, lbls = build_masked(sample["conversations"], tok) trainable = sum(1 for x in lbls if x != -100) print(f"nEjemplo tokenizado: {len(ids)} tokens, {trainable} trainable ({100*trainable/len(ids):.1f}%)") think_lens, call_lens, ans_lens =[],[],[]para ex en sub.select(range(min(500, len(sub)))): para t en ex["conversaciones"]: si t["de"] != "gpt": continuar p = parse_assistant(t["valor"]) para th en p["pensamientos"]: think_lens.append(len(th)) para c en p["tool_calls"]: call_lens.append(len(json.dumps(c))) if p["final"]: ans_lens.append(len(p["final"])) plt.figure(figsize=(10,4)) plt.hist([think_lens, call_lens, ans_lens], bins=40, log=True, label=["", "", "respuesta final"], stacked=False) plt.leyenda(); plt.xlabel("caracteres"); plt.title("Distribuciones de longitud (log y)") plt.tight_layout(); plt.show() clase TraceReplayer: def __init__(self, ex): self.ex = ex self.steps =[]pendiente = Ninguno para t en ex["conversaciones"]: if t["from"] == "gpt": si pendiente: self.steps.append(pendiente) pendiente = {"think": parse_assistant(t["value"]), "responses":[]} elif t["from"] == "tool" y pendiente: pendiente["respuestas"].append(parse_tool(t["valor"])) si está pendiente: self.steps.append(pendiente) def __len__(self): return len(self.steps) def play(self, i): s = self.steps[i] print(f"n── Paso {i+1}/{len(self)} ──") para th en s["think"]["thinkments"]: print(f"💭 {textwrap.shorten(th, 280)}") para c en s["think"]["tool_calls"]: print(f"⚙️ {c.get('name')}({json.dumps(c.get('arguments', {}))[:140]})") for r in s["responses"]: print(f"📥 {textwrap.shorten(json.dumps(r), 200)}") if s["think"]["final"]: print(f"💬 {textwrap.shorten(s['think']['final'], 200)}") rp = TraceReplayer(muestra) para i in range(min(3, len(rp))): rp.play(i) TRAIN = False if TRAIN: importar antorcha desde transformadores importar AutoModelForCausalLM desde trl importar SFTTrainer, SFTConfig train_subset = ds.select(range(200)) def to_text(batch): msgs = to_openai_messages(batch["conversations"]) for m in msgs: if m["rol"] == "herramienta": m["rol"] = "usuario"; m["content"] = "[HERRAMIENTA]n" + m["content"] lote["text"] = tok.apply_chat_template(msgs, tokenize=False, add_generación_prompt=False) return lote train_subset = train_subset.map(to_text) model = AutoModelForCausalLM.from_pretrained( TOK_ID, torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32, device_map="auto" if torch.cuda.is_available() else Ninguno, ) cfg = SFTConfig(output_dir="hermes-sft-demo", per_device_train_batch_size=1, gradient_accumulation_steps=4, max_steps=20, learning_rate=2e-5, logging_steps=2, max_seq_length=1024, dataset_text_field="text", report_to="none", fp16=torch.cuda.is_available(), ) SFTTrainer(model=model, args=cfg, train_dataset=train_subset, Processing_class=tok).train() print("Demostración de ajuste finalizada.") print("n✅ Tutorial completo. Ahora tiene analizadores, análisis, gráficos y un " "ejemplos SFT tokenizados + enmascarados con etiquetas y un gancho de entrenamiento opcional.")
Tokenizamos las conversaciones y aplicamos enmascaramiento de etiquetas para que solo las respuestas del asistente contribuyan a la capacitación. Analizamos las distribuciones de longitud del razonamiento, las llamadas a herramientas y las respuestas para obtener más información. También implementamos un reproductor de seguimiento para recorrer el comportamiento del agente y, opcionalmente, ejecutar un pequeño ciclo de ajuste.
En conclusión, desarrollamos un flujo de trabajo estructurado para analizar, analizar y trabajar eficazmente con los rastros de razonamiento de los agentes. Pudimos dividir las conversaciones en componentes significativos, examinar cómo los agentes razonan paso a paso y medir cómo interactúan con las herramientas durante la resolución de problemas. Utilizando las visualizaciones y los análisis, obtuvimos información sobre patrones y comportamientos comunes en todo el conjunto de datos. Además, convertimos los datos a un formato adecuado para entrenar modelos de lenguaje, incluido el manejo de tokenización y enmascaramiento de etiquetas para las respuestas del asistente. Además, este proceso proporciona una base sólida para estudiar, evaluar y mejorar los sistemas de IA que utilizan herramientas de una manera práctica y escalable.
Consulte los códigos completos con Notebook. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 130.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.
¿Necesita asociarse con nosotros para promocionar su repositorio de GitHub O su página principal de Hugging O su lanzamiento de producto O seminario web, etc.? Conéctate con nosotros
