Spaces:
Running
Running
File size: 1,910 Bytes
021e065 | 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 | import sys, json
sys.path.insert(0, 'senti')
from sqlalchemy import text
from backend.database.postgres.db import SessionLocal
from backend.database.postgres.models import DomainMemory
from core.brain.memory.manager import MemoryManager
from core.brain.memory.domains.registry import resolve_domains
db = SessionLocal()
user_hash = "test_multidomain_user"
try:
# Clean old records
db.execute(text("DELETE FROM domain_memory WHERE user_hash = :h"), {"h": user_hash})
db.commit()
tax_facts = {
"tax:known_income_kes": {"value": 95000.0, "updated_at": "2026-05-31T00:00:00", "confidence": 1.0, "conflict": False, "historical_values": []},
"tax:tax_regime": {"value": "PAYE", "updated_at": "2026-05-31T00:00:00", "confidence": 1.0, "conflict": False, "historical_values": []}
}
db.add(DomainMemory(user_hash=user_hash, domain="tax", facts_json=json.dumps(tax_facts)))
law_facts = {
"law:jurisdiction": {"value": "KE", "updated_at": "2026-05-31T00:00:00", "confidence": 1.0, "conflict": False, "historical_values": []},
"law:aml_required": {"value": True, "updated_at": "2026-05-31T00:00:00", "confidence": 1.0, "conflict": False, "historical_values": []}
}
db.add(DomainMemory(user_hash=user_hash, domain="law", facts_json=json.dumps(law_facts)))
db.commit()
query = "Explain my legal tax compliance and budget in Kenya"
intent = "PLANNING"
print("Resolved domains:")
active = resolve_domains(query, intent)
for p in active:
print(f" - {p.domain_id} (priority {p.priority})")
manager = MemoryManager(user_hash=user_hash, db=db)
ctx = manager.before_query(query, intent=intent)
print("\n--- Context output: ---")
print(ctx)
print("-----------------------")
finally:
db.execute(text("DELETE FROM domain_memory WHERE user_hash = :h"), {"h": user_hash})
db.commit()
db.close()
|