Hierarchical Adaptive Cognitive Agent System

HACAS is a lightweight research prototype for hierarchical distributed cognition. It is built for offline notebook/runtime environments where the goal is orchestration, adaptation, memory, and reflective correction rather than giant model training.

What It Implements

• Parent orchestrator node for task classification, decomposition, sparse activation, aggregation, and reward updates.
• Specialized root sandbox agents: reasoning, pattern, planning, memory, and critic.
• Shared cognitive memory with FAISS support when available and deterministic numpy fallback when not.
• Solver -> critic -> refiner reflection loop.
• Multi-armed-bandit-inspired adaptive routing.
• Sparse cognitive graph for selective inter-agent communication.
• Async execution pipeline.
• Benchmark and demo scripts.
• Offline ARC neuro-symbolic solver with object extraction, DSL primitives, search, verification, and Kaggle submission generation.

Quick Start

Run the interactive program:

python demos/run_demo.py

When the prompt appears, type your question or task and press Enter.

You can also pass the question on the command line:

python demos/run_demo.py "Find the next value in 2 4 8 16"

Other useful commands:

python benchmarks/evaluate.py
pytest

ARC Prize submission flow:

python -m benchmarks.arc3_submission --data-root path/to/arc-data --output submission.json

The submission writer emits the required two predictions per test input and keeps the output JSON in the expected Kaggle format. A Kaggle-ready notebook is available at demos/arc3_kaggle_notebook.ipynb, and an ablation runner is available at scripts/arc_ablation.py.

The default system uses deterministic offline heuristics and numpy. Optional FAISS, Transformers, Ray, or Stable-Baselines3 can be preloaded into the notebook image and integrated without changing the public API.

If pytest is not available, use the standard-library test runner:

python -m unittest discover tests

Minimal Integration

from hacas import CognitiveOrchestrator, HacasConfig

system = CognitiveOrchestrator(HacasConfig(memory_path=None))
response = system.solve("Plan how to integrate HACAS into my Python backend.")

print(response.answer)
print(response.activated_agents)
print(response.confidence)

Service Integration Pattern

Wrap the orchestrator behind your existing API boundary:

from fastapi import FastAPI
from pydantic import BaseModel
from hacas import CognitiveOrchestrator, HacasConfig

app = FastAPI()
system = CognitiveOrchestrator(HacasConfig(memory_path=None))

class Request(BaseModel):
    task: str

@app.post("/cognition/solve")
async def solve(req: Request):
    response = await system.solve_async(req.task)
    return {
        "answer": response.answer,
        "confidence": response.confidence,
        "agents": response.activated_agents,
        "metrics": response.metrics,
    }

For production use, keep one orchestrator instance per worker process so memory, routing statistics, and agent state can accumulate.

See demos/service_api.py for a FastAPI wrapper.

Asking Questions

For general questions, use python demos/run_demo.py and enter a natural-language prompt such as:

Why should sparse activation improve compute efficiency?

For pattern questions, phrase the task clearly so the pattern agent can route it correctly:

Find the next value in 2 4 8 16

Examples:

Plan how to integrate HACAS into my Python backend.
What is the next value in 1, 1, 2, 3, 5, 8?
Why should memory improve adaptive reasoning?

If you are running the API service, use the dedicated endpoint:

POST /patterns/solve

With a string pattern:

{
  "pattern": "1, 1, 2, 3, 5, 8",
  "question": "Find the next value."
}

Or with a numeric sequence:

{
  "sequence": [2, 4, 8, 16],
  "question": "Find the next value."
}

The current detector ensemble covers common user-entered patterns: arithmetic, geometric, Fibonacci-like recurrence, finite-difference polynomial sequences, alternating arithmetic subsequences, cycles, squares, cubes, primes, alphabetic progressions, symbol cycles, and simple grid invariants.

Folder Structure

hacas/
  agents/              Specialized sandbox agents
  arc/                 Offline ARC object parser, DSL, search, verifier, and solver
  memory/              Shared semantic and episodic memory
  routing/             Multi-armed bandit routing policy
  graph/               Sparse cognitive communication graph
  evaluation/          Metrics helpers
  orchestrator.py      Parent executive node
  reflection.py        Solver-critic-refiner loop
  task_classifier.py   Lightweight task classifier
demos/                 Runnable sample tasks
benchmarks/            Evaluation harness
docs/                  Research architecture and integration notes
tests/                 Smoke tests
scripts/               CLI utilities including ARC ablation and inspection tools

Runtime Budget

The default prototype runs in seconds on CPU. Under a 9-hour CPU/GPU notebook budget, the recommended research usage is to preload small open-source models and datasets, then keep all training disabled or restricted to routing statistics and shallow adapters.