async def demo_memory(): explicar( “DEMO 4 — Memoria: MEMORIA.md persistente entre sesiones”, “””La memoria a largo plazo sobrevive entre ejecuciones al persistir en MEMORIA.md. En la sesión 1, el agente registra una preferencia del usuario; en una sesión 2 completamente nueva (motor nuevo, transcripción nueva) esa memoria se inyecta en el indicador del sistema, por lo que el agente ya ‘conoce’ al usuario.”””) mem_path = os.path.join(tempfile.gettempdir(), “oh_demo4_MEMORY.md”) memoria = MemoryStore(mem_path) memoria.reset() registro = build_registry() print(” ── Sesión 1 ──”) ctx1 = ToolContext(vfs=VirtualFS(), memoria=memoria, skills=SkillLibrary()) s1 = [
Use(“I’ll remember the user’s stated preferences.”,
[(“remember”, {“note”: “User prefers metric units and concise answers.”})]), lambda m: Say(“Anotó sus preferencias para la próxima vez.”), ]eng1 = QueryEngine(brain=ScriptedBrain(s1), registro=registry, ctx=ctx1, perms=PermissionChecker(PermissionMode.AUTO), hooks=HookManager(), system_prompt=assemble_system_prompt( base=BASE_SYSTEM, project_context=””, memoria=memoria.read(), skills_summary=”(none)”, tool_names=registry.names())) await eng1.run(“Recuerda que me gustan las unidades métricas y las respuestas cortas.”) print(f” MEMORY.md ahora es:\n{textwrap.indent(memory.read(), ‘ ‘)}”) print(“\n ── Sesión 2 (nueva sesión, memoria recargada desde el disco) ──”) memoria2 = MemoryStore(mem_path) ctx2 = ToolContext(vfs=VirtualFS(), memoria=memoria2, habilidades=SkillLibrary()) sysprompt2 = ensamblar_sistema_prompt( base=BASE_SYSTEM, project_context=””, memoria=memoria2.read(), habilidades_summary=”(none)”, nombres_herramientas=registry.names()) print(” El nuevo El mensaje del sistema ya contiene:”) print(textwrap.indent(“## Memoria a largo plazo (MEMORY.md)\n” + memoria2.read(), ” “)) s2 = [lambda m: Say(“Since you prefer metric and brevity: it’s about 5 km. 🙂”)]
eng2 = QueryEngine(brain=ScriptedBrain(s2), registro=registro, ctx=ctx2, perms=PermissionChecker(PermissionMode.AUTO), hooks=HookManager(), system_prompt=sysprompt2) final = await eng2.run(“¿Qué tan lejos es una carrera de 5000 metros, aproximadamente?”) print(f”\n FINAL: {final}”) print(“\n TAKEAWAY: el estado que debería sobrevivir a una conversación va a la memoria, ” “luego se reinyecta al inicio de futuras sesiones.”) async def demo_compaction(): explicar( “DEMO 5 — Autocompactación de contexto (sesiones de varios días sin desbordamiento)”, “””A medida que una sesión crece, la transcripción puede pasar la ventana de contexto. La autocompactación resume la mitad anterior de la conversación en una nota compacta mientras se preserva la tarea original y la mayoría giros recientes, por lo que los agentes de larga duración continúan (forzamos un pequeño umbral para activarlo; OpenHarness real le pide al modelo que escriba el resumen).”””) msgs = [Message(role=”user”, content=”Build and verify a data pipeline.”)]
para i en el rango(8): msgs.append(Message(role=”assistant”, content=f”Paso {i}: trabajando…”, tool_calls=[ToolCall(f”c{i}”, “shell”,
{“command”: f”process chunk {i}”})])) msgs.append(Message(role=”tool”, name=”shell”, tool_call_id=f”c{i}”, content=f”chunk {i} procesado: 1000 filas ok ” * 4)) before = estima_messages_tokens(msgs) print(f” Antes: {len(msgs)} mensajes, ~{before} tokens”) compactado = may_compact(msgs, max_tokens=300, keep_last=4) after = estimate_messages_tokens(compacted) print(f” Después: {len(compacted)} mensajes, ~{after} tokens ” f”({100 * (antes – después) // antes}% más pequeño)”) print(“\n El mensaje de resumen inyectado:”) print(textwrap.indent(compacted[1].content, ” “)) print(“\n CONSEJO: el arnés administra la ventana de contexto para que el ” “agente pueda ejecutarse mucho más tiempo del que permite una sola ventana.”) async def demo_multi_agent(): explicar( “DEMO 6 — Coordinación de enjambre: generación de subagentes paralelos”, “””Un agente líder descompone una tarea y la delega a subagentes especializados. Cada subagente es su PROPIO arnés (bucle propio, cerebro propio, herramientas propias). Dos investigadores ejecutan EN PARALELO (publicado en el mismo turno → asyncio.gather), luego un escritor sintetiza sus hallazgos. El registro del equipo rastrea quién hizo qué.”””) def researcher_profile(): reg = build_registry(.[WebSearchTool]) guión = [
Use(“Researching via web search.”,
[(“web_search”, {“query”: “PLACEHOLDER”})]), lambda m: Say(“Resumen: ” + short(last_tool_results(m)[0][“content”]160)), ]devuelve ScriptedBrain(script), reg def escritor_perfil(): reg = build_registry([WriteFileTool]) guión = [lambda m: Say(“Synthesized brief combining both research notes ”
“into a coherent paragraph.”)]
return ScriptedBrain(script), reg perfiles = {“researcher”: investigador_perfil, “escritor”: escritor_perfil} vfs = VirtualFS() memoria = MemoryStore(os.path.join(tempfile.gettempdir(), “oh_d6.md”)) habilidades = SkillLibrary() equipo: lista = []
def make_spawn(): async def spawn(role: str, tarea: str) -> str: factory = perfiles.get(role) if not factory: return f”(no existe tal rol: {role})” child_brain, child_reg = factory() if role == “investigador” y child_brain.script: child_brain.script[0] = Usar(f”Investigando: {tarea}”,
[(“web_search”, {“query”: task})]) child_ctx = ToolContext(vfs=vfs, memoria=memoria, habilidades=habilidades, spawn=spawn) child_engine = QueryEngine( cerebro=child_brain, registro=child_reg, ctx=child_ctx, perms=PermissionChecker(PermissionMode.AUTO), ganchos=HookManager(), system_prompt=”(subagent)”, aprobar=auto_approve, max_turns=6) print(f” 🧑🔧 generado [{role}] para: {short(task, 60)}”) resultado = await child_engine.run(task, on_event=None) team.append({“role”: role, “task”: task, “result”: result}) devolver resultado return spawn ctx = ToolContext(vfs=vfs, memoria=memoria, habilidades=habilidades, spawn=make_spawn()) registro = build_registry() lead_script = [
Use(“I’ll split this: research vector databases AND agent harnesses in ”
“parallel, then have a writer combine the findings.”,
[(“spawn_agent”, {“role”: “researcher”,
“task”: “vector database for RAG”}),
(“spawn_agent”, {“role”: “researcher”,
“task”: “agent harness design”})]), Use(“Ambas notas de investigación están en – delegando la síntesis al escritor.”,
[(“spawn_agent”, {“role”: “writer”,
“task”: “combine the two research notes”})]), lambda m: Say(“Coordinación completa: 2 investigadores (paralelos) + 1 ” “escritor produjo un informe combinado.”), ]motor = QueryEngine(brain=ScriptedBrain(lead_script), registro=registro, ctx=ctx, perms=PermissionChecker(PermissionMode.AUTO), ganchos=HookManager(), system_prompt=”(agente principal)”, max_turns=8) imprimir(“\n[running the lead agent]\n”) t0 = time.time() final = await Engine.run(“Elabore un breve resumen sobre la creación de agentes RAG.”) dt = time.time() – t0 print(f”\n FINAL: {final}”) print(f”\n Registro del equipo ({len(team)} ejecuciones de subagente, total {dt:.3f}s):”) para ingresar al equipo: print(f” – [{entry[‘role’]}]{corto(entrada[‘task’]40)} -> ” f”{corto(entrada[‘result’]80)}”) print(“\n TAKEAWAY: el mismo bucle se anida; una ‘herramienta’ puede ser un agente completo, ” “lo que permite la delegación y equipos paralelos.”) async def demo_real_provider(): explicar( “DEMO 7: Intercambio en un modelo REAL (antrópico / compatible con OpenAI)”, “””Todo lo anterior se ejecutó en un cerebro simulado determinista: cero claves, costo cero. La puesta en marcha cambia exactamente UNA cosa: el El motor, las herramientas, los permisos, los enlaces, las habilidades, la memoria y el coordinador están intactos. Este es el objetivo de un arnés: el modelo es conectable.”””) print(textwrap.dedent(“””\ Para ejecutar el MISMO arnés en un modelo real, configure las variables de entorno y vuelva a ejecutarlo (funciona con cualquier punto final compatible con OpenAI o Anthropic que admita OpenHarness: Claude, GPT, Kimi, GLM, DeepSeek, Qwen, Groq, Ollama, OpenRouter, …): importar os.environ[“USE_REAL_LLM”] = “1” # — Estilo antrópico — os.environ[“ANTHROPIC_API_KEY”] = “sk-hormiga-…” os.environ[“MODEL”] = “claude-sonnet-4-6” # — o estilo OpenAI (incluido Ollama local) — # os.environ[“OPENAI_API_KEY”] = “sk-…” # os.environ[“OPENAI_BASE_URL”] = “http://localhost:11434/v1” # os.environ[“MODEL”] = “llama-3.3-70b” Luego construye el motor con el cerebro real en lugar del simulacro: cerebro = make_real_brain(system=system_prompt) o ScriptedBrain([…]) motor = QueryEngine(cerebro=cerebro, registro=registro, ctx=ctx, …) aguarda motor.run(“Refactor utils.py y agrega pruebas.”) “””)) sysprompt = ensamblar_sistema_prompt( base=BASE_SYSTEM, project_context=””, memoria=””, skills_summary=”(none)”, nombres_herramientas=build_registry().names()) real = make_real_brain(system=sysprompt) si real es Ninguno: print(” [USE_REAL_LLM not set → staying on the mock brain. ”
“Set the env vars above and re-run to go live.]”) devolver imprimir(f” [LIVE] Usando proveedor real: {real.api_format} / {real.model}\n”) vfs = VirtualFS() ctx = ToolContext(vfs=vfs, Memory=MemoryStore( os.path.join(tempfile.gettempdir(), “oh_real.md”)), skills=SkillLibrary(), canned_answers={}) motor = QueryEngine( Brain=RetryingBrain(real), registro=build_registry(), ctx=ctx, perms=PermissionChecker(PermissionMode.AUTO), hooks=HookManager(), system_prompt=sysprompt, cost=CostMeter(real.model), max_turns=12) final = await Engine.run( “Crea greet.py con una función greet(name) que devuelve ” “‘¡Hola,!’, luego escribe y ejecuta una prueba rápida para demostrar que funciona.”) print(f”\n FINAL: {final}”) print(f”\n Archivos:\n{vfs.tree()}”) print(f”\n 💰 {engine.cost.summary()}”) async def main(): banner(“OpenHarness From Scratch – tutorial guiado”) print(textwrap.dedent(“”” Construiremos el arnés un subsistema a la vez: 1. El bucle del agente (herramientas, ejecutar/verificar/reparar, reintentos, costo) 2. Permisos (modos, rutas sensibles, reglas, veto de gancho) 3. Habilidades (conocimiento bajo demanda) 4. Memoria (MEMORIA.md persistente entre sesiones) 5. Compactación (sobrevivir a sesiones largas) 6. Multi-agente (delegación de subagente paralelo) 7. Proveedor real (intercambio de una línea a un modelo en vivo) Arquitectura (de qué es responsable cada pieza): Aviso del usuario │ ▼ QueryEngine ── ► LLM Brain (simulado o real) “QUÉ hacer” │ ▲ │ tool_use │ └────────────┘ ▼ Para cada llamada de herramienta: Permiso ─ ► PreHook ─ ► Ejecutar ─ ► PostHook │ │ │ │ negar/pedir veto/editar redacción de sandbox │ ▼ Resultado de la herramienta ── ► volver a la transcripción ── ► bucle “””).rstrip()) en espera demo_agent_loop() en espera demo_permissions() en espera demo_skills() en espera demo… y memoria persistente • autocompactación de contexto • coordinación anidada de múltiples agentes • un intercambio de una línea a un proveedor LLM real Para profundizar, estudie el proyecto real: https://github.com/HKUDS/OpenHarness (más de 43 herramientas, ecosistema de complementos, cliente MCP, React/Ink TUI, la CLI `oh` y el agente personal `ohmo`).
Use(“I’ll remember the user’s stated preferences.”,
[(“remember”, {“note”: “User prefers metric units and concise answers.”})]), lambda m: Say(“Anotó sus preferencias para la próxima vez.”), ]eng1 = QueryEngine(brain=ScriptedBrain(s1), registro=registry, ctx=ctx1, perms=PermissionChecker(PermissionMode.AUTO), hooks=HookManager(), system_prompt=assemble_system_prompt( base=BASE_SYSTEM, project_context=””, memoria=memoria.read(), skills_summary=”(none)”, tool_names=registry.names())) await eng1.run(“Recuerda que me gustan las unidades métricas y las respuestas cortas.”) print(f” MEMORY.md ahora es:\n{textwrap.indent(memory.read(), ‘ ‘)}”) print(“\n ── Sesión 2 (nueva sesión, memoria recargada desde el disco) ──”) memoria2 = MemoryStore(mem_path) ctx2 = ToolContext(vfs=VirtualFS(), memoria=memoria2, habilidades=SkillLibrary()) sysprompt2 = ensamblar_sistema_prompt( base=BASE_SYSTEM, project_context=””, memoria=memoria2.read(), habilidades_summary=”(none)”, nombres_herramientas=registry.names()) print(” El nuevo El mensaje del sistema ya contiene:”) print(textwrap.indent(“## Memoria a largo plazo (MEMORY.md)\n” + memoria2.read(), ” “)) s2 = [lambda m: Say(“Since you prefer metric and brevity: it’s about 5 km. 🙂”)]
eng2 = QueryEngine(brain=ScriptedBrain(s2), registro=registro, ctx=ctx2, perms=PermissionChecker(PermissionMode.AUTO), hooks=HookManager(), system_prompt=sysprompt2) final = await eng2.run(“¿Qué tan lejos es una carrera de 5000 metros, aproximadamente?”) print(f”\n FINAL: {final}”) print(“\n TAKEAWAY: el estado que debería sobrevivir a una conversación va a la memoria, ” “luego se reinyecta al inicio de futuras sesiones.”) async def demo_compaction(): explicar( “DEMO 5 — Autocompactación de contexto (sesiones de varios días sin desbordamiento)”, “””A medida que una sesión crece, la transcripción puede pasar la ventana de contexto. La autocompactación resume la mitad anterior de la conversación en una nota compacta mientras se preserva la tarea original y la mayoría giros recientes, por lo que los agentes de larga duración continúan (forzamos un pequeño umbral para activarlo; OpenHarness real le pide al modelo que escriba el resumen).”””) msgs = [Message(role=”user”, content=”Build and verify a data pipeline.”)]
para i en el rango(8): msgs.append(Message(role=”assistant”, content=f”Paso {i}: trabajando…”, tool_calls=[ToolCall(f”c{i}”, “shell”,
{“command”: f”process chunk {i}”})])) msgs.append(Message(role=”tool”, name=”shell”, tool_call_id=f”c{i}”, content=f”chunk {i} procesado: 1000 filas ok ” * 4)) before = estima_messages_tokens(msgs) print(f” Antes: {len(msgs)} mensajes, ~{before} tokens”) compactado = may_compact(msgs, max_tokens=300, keep_last=4) after = estimate_messages_tokens(compacted) print(f” Después: {len(compacted)} mensajes, ~{after} tokens ” f”({100 * (antes – después) // antes}% más pequeño)”) print(“\n El mensaje de resumen inyectado:”) print(textwrap.indent(compacted[1].content, ” “)) print(“\n CONSEJO: el arnés administra la ventana de contexto para que el ” “agente pueda ejecutarse mucho más tiempo del que permite una sola ventana.”) async def demo_multi_agent(): explicar( “DEMO 6 — Coordinación de enjambre: generación de subagentes paralelos”, “””Un agente líder descompone una tarea y la delega a subagentes especializados. Cada subagente es su PROPIO arnés (bucle propio, cerebro propio, herramientas propias). Dos investigadores ejecutan EN PARALELO (publicado en el mismo turno → asyncio.gather), luego un escritor sintetiza sus hallazgos. El registro del equipo rastrea quién hizo qué.”””) def researcher_profile(): reg = build_registry(.[WebSearchTool]) guión = [
Use(“Researching via web search.”,
[(“web_search”, {“query”: “PLACEHOLDER”})]), lambda m: Say(“Resumen: ” + short(last_tool_results(m)[0][“content”]160)), ]devuelve ScriptedBrain(script), reg def escritor_perfil(): reg = build_registry([WriteFileTool]) guión = [lambda m: Say(“Synthesized brief combining both research notes ”
“into a coherent paragraph.”)]
return ScriptedBrain(script), reg perfiles = {“researcher”: investigador_perfil, “escritor”: escritor_perfil} vfs = VirtualFS() memoria = MemoryStore(os.path.join(tempfile.gettempdir(), “oh_d6.md”)) habilidades = SkillLibrary() equipo: lista = []
def make_spawn(): async def spawn(role: str, tarea: str) -> str: factory = perfiles.get(role) if not factory: return f”(no existe tal rol: {role})” child_brain, child_reg = factory() if role == “investigador” y child_brain.script: child_brain.script[0] = Usar(f”Investigando: {tarea}”,
[(“web_search”, {“query”: task})]) child_ctx = ToolContext(vfs=vfs, memoria=memoria, habilidades=habilidades, spawn=spawn) child_engine = QueryEngine( cerebro=child_brain, registro=child_reg, ctx=child_ctx, perms=PermissionChecker(PermissionMode.AUTO), ganchos=HookManager(), system_prompt=”(subagent)”, aprobar=auto_approve, max_turns=6) print(f” 🧑🔧 generado [{role}] para: {short(task, 60)}”) resultado = await child_engine.run(task, on_event=None) team.append({“role”: role, “task”: task, “result”: result}) devolver resultado return spawn ctx = ToolContext(vfs=vfs, memoria=memoria, habilidades=habilidades, spawn=make_spawn()) registro = build_registry() lead_script = [
Use(“I’ll split this: research vector databases AND agent harnesses in ”
“parallel, then have a writer combine the findings.”,
[(“spawn_agent”, {“role”: “researcher”,
“task”: “vector database for RAG”}),
(“spawn_agent”, {“role”: “researcher”,
“task”: “agent harness design”})]), Use(“Ambas notas de investigación están en – delegando la síntesis al escritor.”,
[(“spawn_agent”, {“role”: “writer”,
“task”: “combine the two research notes”})]), lambda m: Say(“Coordinación completa: 2 investigadores (paralelos) + 1 ” “escritor produjo un informe combinado.”), ]motor = QueryEngine(brain=ScriptedBrain(lead_script), registro=registro, ctx=ctx, perms=PermissionChecker(PermissionMode.AUTO), ganchos=HookManager(), system_prompt=”(agente principal)”, max_turns=8) imprimir(“\n[running the lead agent]\n”) t0 = time.time() final = await Engine.run(“Elabore un breve resumen sobre la creación de agentes RAG.”) dt = time.time() – t0 print(f”\n FINAL: {final}”) print(f”\n Registro del equipo ({len(team)} ejecuciones de subagente, total {dt:.3f}s):”) para ingresar al equipo: print(f” – [{entry[‘role’]}]{corto(entrada[‘task’]40)} -> ” f”{corto(entrada[‘result’]80)}”) print(“\n TAKEAWAY: el mismo bucle se anida; una ‘herramienta’ puede ser un agente completo, ” “lo que permite la delegación y equipos paralelos.”) async def demo_real_provider(): explicar( “DEMO 7: Intercambio en un modelo REAL (antrópico / compatible con OpenAI)”, “””Todo lo anterior se ejecutó en un cerebro simulado determinista: cero claves, costo cero. La puesta en marcha cambia exactamente UNA cosa: el El motor, las herramientas, los permisos, los enlaces, las habilidades, la memoria y el coordinador están intactos. Este es el objetivo de un arnés: el modelo es conectable.”””) print(textwrap.dedent(“””\ Para ejecutar el MISMO arnés en un modelo real, configure las variables de entorno y vuelva a ejecutarlo (funciona con cualquier punto final compatible con OpenAI o Anthropic que admita OpenHarness: Claude, GPT, Kimi, GLM, DeepSeek, Qwen, Groq, Ollama, OpenRouter, …): importar os.environ[“USE_REAL_LLM”] = “1” # — Estilo antrópico — os.environ[“ANTHROPIC_API_KEY”] = “sk-hormiga-…” os.environ[“MODEL”] = “claude-sonnet-4-6” # — o estilo OpenAI (incluido Ollama local) — # os.environ[“OPENAI_API_KEY”] = “sk-…” # os.environ[“OPENAI_BASE_URL”] = “http://localhost:11434/v1” # os.environ[“MODEL”] = “llama-3.3-70b” Luego construye el motor con el cerebro real en lugar del simulacro: cerebro = make_real_brain(system=system_prompt) o ScriptedBrain([…]) motor = QueryEngine(cerebro=cerebro, registro=registro, ctx=ctx, …) aguarda motor.run(“Refactor utils.py y agrega pruebas.”) “””)) sysprompt = ensamblar_sistema_prompt( base=BASE_SYSTEM, project_context=””, memoria=””, skills_summary=”(none)”, nombres_herramientas=build_registry().names()) real = make_real_brain(system=sysprompt) si real es Ninguno: print(” [USE_REAL_LLM not set → staying on the mock brain. ”
“Set the env vars above and re-run to go live.]”) devolver imprimir(f” [LIVE] Usando proveedor real: {real.api_format} / {real.model}\n”) vfs = VirtualFS() ctx = ToolContext(vfs=vfs, Memory=MemoryStore( os.path.join(tempfile.gettempdir(), “oh_real.md”)), skills=SkillLibrary(), canned_answers={}) motor = QueryEngine( Brain=RetryingBrain(real), registro=build_registry(), ctx=ctx, perms=PermissionChecker(PermissionMode.AUTO), hooks=HookManager(), system_prompt=sysprompt, cost=CostMeter(real.model), max_turns=12) final = await Engine.run( “Crea greet.py con una función greet(name) que devuelve ” “‘¡Hola,!’, luego escribe y ejecuta una prueba rápida para demostrar que funciona.”) print(f”\n FINAL: {final}”) print(f”\n Archivos:\n{vfs.tree()}”) print(f”\n 💰 {engine.cost.summary()}”) async def main(): banner(“OpenHarness From Scratch – tutorial guiado”) print(textwrap.dedent(“”” Construiremos el arnés un subsistema a la vez: 1. El bucle del agente (herramientas, ejecutar/verificar/reparar, reintentos, costo) 2. Permisos (modos, rutas sensibles, reglas, veto de gancho) 3. Habilidades (conocimiento bajo demanda) 4. Memoria (MEMORIA.md persistente entre sesiones) 5. Compactación (sobrevivir a sesiones largas) 6. Multi-agente (delegación de subagente paralelo) 7. Proveedor real (intercambio de una línea a un modelo en vivo) Arquitectura (de qué es responsable cada pieza): Aviso del usuario │ ▼ QueryEngine ── ► LLM Brain (simulado o real) “QUÉ hacer” │ ▲ │ tool_use │ └────────────┘ ▼ Para cada llamada de herramienta: Permiso ─ ► PreHook ─ ► Ejecutar ─ ► PostHook │ │ │ │ negar/pedir veto/editar redacción de sandbox │ ▼ Resultado de la herramienta ── ► volver a la transcripción ── ► bucle “””).rstrip()) en espera demo_agent_loop() en espera demo_permissions() en espera demo_skills() en espera demo… y memoria persistente • autocompactación de contexto • coordinación anidada de múltiples agentes • un intercambio de una línea a un proveedor LLM real Para profundizar, estudie el proyecto real: https://github.com/HKUDS/OpenHarness (más de 43 herramientas, ecosistema de complementos, cliente MCP, React/Ink TUI, la CLI `oh` y el agente personal `ohmo`).