Spaces:
Running on Zero
Running on Zero
File size: 2,726 Bytes
9fb722e d4978df 9fb722e d4978df 9fb722e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | from __future__ import annotations
import asyncio
from typing import Any
from hearthnet.controller import HearthNetController
from hearthnet.node import HearthNode, InMemoryNetwork
__all__ = [
"HearthNetController",
"HearthNode",
"InMemoryNetwork",
"answer",
"get_capabilities",
"status",
]
def _build_test_network() -> InMemoryNetwork:
"""Build a minimal in-memory network for unit tests and CI only.
Uses echo-LLM services (not real models). Never call from production paths.
"""
network = InMemoryNetwork()
anchor = network.add_node("ed25519:anchor", "Anchor Workstation")
hearth = network.add_node("ed25519:hearth", "Hearth Laptop")
spark = network.add_node("ed25519:spark", "Spark Phone")
anchor.install_demo_services(corpus="niederrhein-demo")
hearth.install_demo_services()
spark.install_demo_services()
network.mesh_discover()
return network
# Keep _build_demo_network as alias so existing tests don't break
_build_demo_network = _build_test_network
def get_capabilities() -> dict[str, Any]:
network = _build_demo_network()
anchor = network.nodes[0]
snapshot = anchor.snapshot()
return {
"system_of_concern": "community-owned resilient AI assistance",
"controller": "HearthNetController -> HearthNode",
"facades": ["RagFacade", "ChatFacade", "MarketplaceFacade"],
"bus": "CapabilityBus",
"local": snapshot["topology"].capabilities_local,
"remote": snapshot["topology"].capabilities_remote,
}
def status() -> dict[str, Any]:
network = _build_demo_network()
anchor = network.nodes[0]
snapshot = anchor.snapshot()
return {
"node": snapshot["node"],
"peers": snapshot["topology"].peers,
"emergency": snapshot["emergency"].mode,
}
def answer(question: str) -> str:
network = _build_demo_network()
anchor = network.nodes[0]
result = asyncio.run(
anchor.bus.call(
"rag.query",
(1, 0),
{
"params": {"corpus": "niederrhein-demo"},
"input": {"query": question, "k": 2},
},
)
)
chunks = result["output"]["chunks"]
if not chunks:
return "No local chunk matched. The bus stayed local and returned an auditable miss."
source_lines = "\n".join(
f"- {chunk['metadata']['doc_title']}: {chunk['text']}" for chunk in chunks
)
return (
"HearthNet routed this through `rag.query@1.0` on the local capability bus.\n\n"
f"Local context:\n{source_lines}\n\n"
"Phase 1 keeps the demo deterministic while preserving the controller -> facade -> bus -> service shape."
)
|