64b800c6cf
Items completed: 1. Merged PR #2 (starlette/httpx deps) 2. Fixed async race condition in multimodal_ui.py 3. Wired TTSAdapter (ElevenLabs, Azure) in API routes 4. Moved super_big_brain.py from core/ to reasoning/ (backward compat shim) 5. Added API authentication middleware (Bearer token via FUSIONAGI_API_KEY) 6. Added async adapter interface (acomplete/acomplete_structured) 7. Migrated FastAPI on_event to lifespan (fixes 20 deprecation warnings) 8. Liquid Neural Networks (continuous-time adaptive weights) 9. Quantum-AI Hybrid compute backend (simulator + optimization) 10. Embodied Intelligence / Robotics bridge (actuator + sensor protocols) 11. Consciousness Engineering (formal self-model with introspection) 12. ASI Scoring Rubric (C/A/L/N/R self-assessment harness) 13. GPU integration tests for TensorFlow backend 14. Multi-stage production Dockerfile 15. Gitea CI/CD pipeline (lint, test matrix, Docker build) 16. API rate limiting middleware (per-IP sliding window) 17. OpenAPI docs cleanup (auth + rate limiting descriptions) 18. Benchmarking suite (decomposition, multi-path, recomposition, e2e) 19. Plugin system (head registry for custom heads) 427 tests passing, 0 ruff errors, 0 mypy errors. Co-Authored-By: Nakamoto, S <defi@defi-oracle.io>
131 lines
4.2 KiB
Python
131 lines
4.2 KiB
Python
"""Tests for Dvādaśa 12-head FusionAGI components."""
|
|
|
|
|
|
from fusionagi import EventBus, Orchestrator, StateManager
|
|
from fusionagi.adapters import StubAdapter
|
|
from fusionagi.agents import WitnessAgent
|
|
from fusionagi.agents.heads import create_all_content_heads
|
|
from fusionagi.core import run_dvadasa, run_heads_parallel, select_heads_for_complexity
|
|
from fusionagi.multi_agent import run_consensus
|
|
from fusionagi.schemas import (
|
|
HeadClaim,
|
|
HeadId,
|
|
HeadOutput,
|
|
UserIntent,
|
|
parse_user_input,
|
|
)
|
|
|
|
|
|
def test_parse_user_input_normal():
|
|
cmd = parse_user_input("What is the best approach?")
|
|
assert cmd.intent == UserIntent.NORMAL
|
|
assert cmd.cleaned_prompt == "What is the best approach?"
|
|
|
|
|
|
def test_parse_user_input_head_strategy():
|
|
cmd = parse_user_input("/head strategy What is the best approach?")
|
|
assert cmd.intent == UserIntent.HEAD_STRATEGY
|
|
assert cmd.head_id == HeadId.STRATEGY
|
|
assert "best approach" in cmd.cleaned_prompt
|
|
|
|
|
|
def test_parse_user_input_show_dissent():
|
|
cmd = parse_user_input("/show dissent")
|
|
assert cmd.intent == UserIntent.SHOW_DISSENT
|
|
|
|
|
|
def test_head_output_schema():
|
|
out = HeadOutput(
|
|
head_id=HeadId.LOGIC,
|
|
summary="Test",
|
|
claims=[
|
|
HeadClaim(claim_text="X is true", confidence=0.9, evidence=[], assumptions=[]),
|
|
],
|
|
risks=[],
|
|
questions=[],
|
|
recommended_actions=[],
|
|
tone_guidance="",
|
|
)
|
|
assert out.head_id == HeadId.LOGIC
|
|
assert len(out.claims) == 1
|
|
assert out.claims[0].confidence == 0.9
|
|
|
|
|
|
def test_consensus_engine():
|
|
outputs = [
|
|
HeadOutput(
|
|
head_id=HeadId.LOGIC,
|
|
summary="S1",
|
|
claims=[
|
|
HeadClaim(claim_text="X is true", confidence=0.8, evidence=[], assumptions=[]),
|
|
],
|
|
risks=[],
|
|
questions=[],
|
|
recommended_actions=[],
|
|
tone_guidance="",
|
|
),
|
|
]
|
|
am = run_consensus(outputs)
|
|
assert am.confidence_score >= 0
|
|
assert isinstance(am.agreed_claims, list)
|
|
assert isinstance(am.disputed_claims, list)
|
|
|
|
|
|
def test_create_all_heads():
|
|
heads = create_all_content_heads()
|
|
assert len(heads) == 11
|
|
assert HeadId.WITNESS not in heads
|
|
|
|
|
|
def test_run_heads_parallel():
|
|
stub = StubAdapter(structured_response={
|
|
"head_id": "logic",
|
|
"summary": "Stub",
|
|
"claims": [],
|
|
"risks": [],
|
|
"questions": [],
|
|
"recommended_actions": [],
|
|
"tone_guidance": "",
|
|
})
|
|
bus = EventBus()
|
|
state = StateManager()
|
|
orch = Orchestrator(event_bus=bus, state_manager=state)
|
|
heads = create_all_content_heads(adapter=stub)
|
|
for hid, agent in list(heads.items())[:2]:
|
|
orch.register_agent(hid.value, agent)
|
|
orch.register_agent(HeadId.WITNESS.value, WitnessAgent(adapter=stub))
|
|
|
|
task_id = orch.submit_task(goal="Test")
|
|
results = run_heads_parallel(orch, task_id, "Hello", head_ids=[HeadId.LOGIC, HeadId.RESEARCH])
|
|
assert len(results) >= 1
|
|
assert all(isinstance(r, HeadOutput) for r in results)
|
|
|
|
|
|
def test_select_heads_for_complexity():
|
|
simple = select_heads_for_complexity("What is 2+2?")
|
|
assert len(simple) <= 5
|
|
complex_heads = select_heads_for_complexity(
|
|
"We need to design a secure architecture for production with compliance requirements"
|
|
)
|
|
assert len(complex_heads) == 11
|
|
|
|
|
|
def test_run_dvadasa_native_reasoning():
|
|
"""Test Dvādaśa runs with native reasoning (no external LLM)."""
|
|
bus = EventBus()
|
|
state = StateManager()
|
|
orch = Orchestrator(event_bus=bus, state_manager=state)
|
|
# adapter=None => uses NativeReasoningProvider for heads, NativeAdapter for Witness
|
|
heads = create_all_content_heads(adapter=None)
|
|
for hid, agent in list(heads.items())[:3]: # Just Logic, Research, Systems
|
|
orch.register_agent(hid.value, agent)
|
|
orch.register_agent(HeadId.WITNESS.value, WitnessAgent(adapter=None))
|
|
task_id = orch.submit_task(goal="What is the best approach for secure authentication?")
|
|
final = run_dvadasa(
|
|
orch, task_id, "What is the best approach for secure authentication?", event_bus=bus
|
|
)
|
|
assert final is not None
|
|
assert final.final_answer
|
|
assert len(final.final_answer) > 20
|
|
assert final.confidence_score >= 0
|