semantica/cookbook/integrations/agno_decision_intelligence.ipynb

Agno × Semantica: Decision Intelligence Agent

This notebook shows how to wire Semantica's Decision Intelligence stack into an Agno agent so it can:

• Record every decision it makes with full reasoning provenance
• Search historical precedents before acting
• Validate decisions against policy rules
• Trace causal chains across decisions
• Accumulate institutional knowledge that survives across sessions

Domain used: Financial loan underwriting (easily adapted to healthcare, legal, HR, etc.)

---

Architecture

Agno Agent
  ├── memory=AgnoContextStore       ← graph-backed persistent memory
  └── tools=[AgnoDecisionKit]       ← decision tools the LLM can call
          │
          ├── record_decision       ← Semantica AgentContext.record_decision()
          ├── find_precedents       ← Semantica AgentContext.find_precedents_advanced()
          ├── trace_causal_chain    ← Semantica ContextGraph.trace_decision_causality()
          ├── analyze_impact        ← Semantica AgentContext.analyze_decision_influence()
          ├── check_policy          ← Semantica PolicyEngine
          └── get_decision_summary  ← Semantica AgentContext.get_context_insights()

Install

pip install semantica[agno]

1. Setup — Semantica Backends

We build the Semantica components first. These are independent of Agno — you can swap backends without touching agent code.

In [ ]:

import sys, os
sys.path.insert(0, os.path.abspath("../../"))

# ── Semantica core (not Agno-specific) ──────────────────────────────────────
from semantica.context import AgentContext, ContextGraph
from semantica.context import PolicyEngine, DecisionQuery, CausalChainAnalyzer
from semantica.vector_store import VectorStore

# ── Agno integration layer ───────────────────────────────────────────────────
from integrations.agno import AgnoContextStore, AgnoDecisionKit, AGNO_AVAILABLE

print(f"Semantica imports OK")
print(f"Agno installed: {AGNO_AVAILABLE}")

In [ ]:

# ── Vector store (FAISS, no external service needed) ────────────────────────
vector_store = VectorStore(backend="faiss", dimension=768)
print("VectorStore ready (FAISS)")

# ── In-memory context graph with full analytics ──────────────────────────────
knowledge_graph = ContextGraph(
    advanced_analytics=True,
    # Switch to neo4j for production:
    # backend="neo4j", uri="bolt://localhost:7687"
)
print("ContextGraph ready (in-memory)")

2. Seed Historical Decisions

Before the agent runs, we pre-load historical decisions using native Semantica APIs so the precedent database is warm.

In production you would ingest from a database or a prior session's graph export.

In [ ]:

# Build a pure-Semantica AgentContext for seeding historical data
seed_context = AgentContext(
    vector_store=vector_store,
    knowledge_graph=knowledge_graph,
    decision_tracking=True,
)

historical_loans = [
    dict(
        category="loan_approval",
        scenario="Applicant: credit score 740, income $95k, DTI 28%, down payment 20%",
        reasoning="Strong credit history, debt load well below 35% threshold, adequate down payment",
        outcome="approved",
        confidence=0.96,
    ),
    dict(
        category="loan_approval",
        scenario="Applicant: credit score 620, income $45k, DTI 42%, down payment 5%",
        reasoning="Credit score below 650 floor, DTI exceeds 40% maximum, insufficient down payment",
        outcome="rejected",
        confidence=0.97,
    ),
    dict(
        category="loan_approval",
        scenario="Applicant: credit score 700, income $72k, DTI 33%, down payment 15%",
        reasoning="Adequate credit, moderate DTI within range, down payment slightly below ideal",
        outcome="approved_with_conditions",
        confidence=0.82,
    ),
    dict(
        category="loan_approval",
        scenario="Applicant: credit score 780, income $130k, DTI 22%, down payment 30%",
        reasoning="Excellent credit, low debt load, strong down payment — low-risk profile",
        outcome="approved",
        confidence=0.99,
    ),
    dict(
        category="loan_approval",
        scenario="Applicant: credit score 660, income $58k, DTI 38%, down payment 10%",
        reasoning="Borderline credit, high DTI, minimal down payment — escalated to senior review",
        outcome="escalated",
        confidence=0.70,
    ),
]

for loan in historical_loans:
    did = seed_context.record_decision(**loan)
    print(f"  Seeded [{loan['outcome']:25s}] → {did}")

print(f"\n{len(historical_loans)} historical decisions loaded into Semantica KG")

3. Define Policy Rules with Semantica

We use PolicyEngine directly — no Agno involvement here. The AgnoDecisionKit.check_policy tool will call this engine during the agent's reasoning loop.

In [ ]:

LENDING_POLICY_RULES = [
    "credit_score >= 650",
    "dti <= 40",
    "down_payment_pct >= 10",
    "confidence >= 0.70",
]

# Verify directly with Semantica's PolicyEngine before wiring to Agno
policy_engine = PolicyEngine(graph_store=knowledge_graph)

test_application = {"credit_score": 720, "dti": 31, "down_payment_pct": 18, "confidence": 0.88}

try:
    result = policy_engine.check_compliance(test_application, LENDING_POLICY_RULES)
    print(f"Policy check result: compliant={getattr(result, 'compliant', 'N/A')}")
    print(f"Violations: {getattr(result, 'violations', [])}")
except Exception as e:
    print(f"PolicyEngine fallback (expected without full rule engine): {e}")

print("\nPolicy rules defined:", LENDING_POLICY_RULES)

4. Build the Agno Decision-Intelligence Agent

Now we wire everything into Agno using the integration classes.

• AgnoContextStore gives the agent persistent graph-backed memory
• AgnoDecisionKit exposes 6 decision tools the LLM can invoke during reasoning

In [ ]:

# ── AgnoContextStore: wraps AgentContext as Agno MemoryDb ────────────────────
store = AgnoContextStore(
    vector_store=vector_store,       # Same store — shares seeded decisions
    knowledge_graph=knowledge_graph, # Same graph — shares seeded decisions
    decision_tracking=True,
    graph_expansion=True,
    session_id="loan_underwriter_v1",
)
print("AgnoContextStore ready")

# ── AgnoDecisionKit: exposes Semantica decision tools to Agno's LLM ──────────
decision_kit = AgnoDecisionKit(
    context=store.context,    # Reuse same AgentContext — shared decision history
    max_precedents=5,
    causal_depth=3,
    enable_policy_check=True,
)
print(f"AgnoDecisionKit ready — {len(decision_kit._tools)} tools registered")
print("  Tools:", [fn.__name__ for fn in decision_kit._tools])

In [ ]:

if AGNO_AVAILABLE:
    from agno.agent import Agent
    from agno.memory import AgentMemory
    from agno.models.openai import OpenAIChat  # or any Agno-supported model

    agent = Agent(
        name="LoanUnderwriter",
        model=OpenAIChat(id="gpt-4o"),
        memory=AgentMemory(db=store),
        tools=[decision_kit],
        show_tool_calls=True,
        description=(
            "You are a senior loan underwriter. Before approving or rejecting any application:"
            " (1) find_precedents for similar past cases,"
            " (2) check_policy compliance,"
            " (3) record_decision with full reasoning."
            " Always cite precedents and policy rule results in your explanation."
        ),
    )
    print("Agno Agent assembled and ready")
else:
    print("Agno not installed — demonstrating tool calls directly below")

5. Demonstrate Decision Tools

We call the decision tools directly so the notebook is fully runnable without an OpenAI key. When Agno is wired, the LLM orchestrates these same calls automatically.

In [ ]:

import json

# ── 5a. Find Precedents ───────────────────────────────────────────────────────
print("=" * 60)
print("TOOL: find_precedents")
print("=" * 60)

new_application_scenario = (
    "Applicant: credit score 715, income $82k, DTI 30%, down payment 18%"
)

precedents_json = decision_kit.find_precedents(
    scenario=new_application_scenario,
    category="loan_approval",
    limit=3,
)
precedents = json.loads(precedents_json)
print(f"Found {precedents['count']} similar past decisions:")
for p in precedents['precedents']:
    print(f"  [{p.get('outcome','?'):25s}] confidence={p.get('confidence',0):.2f}")
    print(f"    {p.get('scenario','')[:80]}")

In [ ]:

# ── 5b. Check Policy ─────────────────────────────────────────────────────────
print("=" * 60)
print("TOOL: check_policy")
print("=" * 60)

decision_data = json.dumps({
    "credit_score": 715,
    "dti": 30,
    "down_payment_pct": 18,
    "confidence": 0.88,
    "outcome": "approved",
})

policy_json = decision_kit.check_policy(
    decision_data=decision_data,
    policy_rules=json.dumps(LENDING_POLICY_RULES),
)
policy_result = json.loads(policy_json)
print(f"Compliant: {policy_result.get('compliant')}")
print(f"Violations: {policy_result.get('violations', [])}")
print(f"Warnings:   {policy_result.get('warnings', [])}")

In [ ]:

# ── 5c. Record Decision ──────────────────────────────────────────────────────
print("=" * 60)
print("TOOL: record_decision")
print("=" * 60)

record_json = decision_kit.record_decision(
    category="loan_approval",
    scenario=new_application_scenario,
    reasoning=(
        "3 similar precedents found — 2 approved, 1 escalated. "
        "Credit score 715 exceeds 650 floor. DTI 30% well within 40% limit. "
        "Down payment 18% above 10% minimum. All policy rules satisfied."
    ),
    outcome="approved",
    confidence=0.91,
    entities="loan_applicant, credit_bureau, lending_policy_v2",
)
record_result = json.loads(record_json)
decision_id = record_result['decision_id']
print(f"Decision recorded: {decision_id}")
print(f"Status: {record_result['status']}")

In [ ]:

# ── 5d. Analyze Impact ───────────────────────────────────────────────────────
print("=" * 60)
print("TOOL: analyze_impact")
print("=" * 60)

impact_json = decision_kit.analyze_impact(decision_id=decision_id)
impact = json.loads(impact_json)
print("Impact analysis:")
for k, v in impact.items():
    if k != "decision_id":
        print(f"  {k}: {v}")

In [ ]:

# ── 5e. Decision Summary ─────────────────────────────────────────────────────
print("=" * 60)
print("TOOL: get_decision_summary")
print("=" * 60)

summary_json = decision_kit.get_decision_summary(category="loan_approval")
summary = json.loads(summary_json)
print("Decision history summary:")
for k, v in summary.items():
    if k not in ("category_filter",):
        print(f"  {k}: {v}")

6. Run the Full Agno Agent (requires API key)

When AGNO_AVAILABLE=True and an OpenAI key is set, the LLM orchestrates all the tool calls automatically.

In [ ]:

NEW_CASE = (
    "New mortgage application received:\n"
    "  Credit score: 715, Annual income: $82,000\n"
    "  Debt-to-income: 30%, Down payment: 18%\n"
    "  Loan amount: $320,000 for a primary residence in Austin TX\n"
    "Should we approve this application?"
)

if AGNO_AVAILABLE:
    agent.print_response(NEW_CASE)
else:
    print("[Agno not installed — skipping live agent run]")
    print()
    print("Expected agent reasoning flow:")
    print("  1. find_precedents('credit score 715, DTI 30%, down payment 18%')")
    print("     → 2 approved, 1 escalated among similar cases")
    print("  2. check_policy(credit_score=715, dti=30, down_payment_pct=18)")
    print("     → compliant=True, violations=[]")
    print("  3. record_decision(outcome='approved', confidence=0.91)")
    print("     → decision_id recorded in Semantica KG")
    print()
    print("  Recommendation: APPROVE — 3 precedents + full policy compliance")

7. Post-Session Analytics with Semantica

After the agent session, use native Semantica APIs for reporting and causal analysis — no Agno required.

In [ ]:

# Query decision history directly from Semantica
insights = store.context.get_context_insights()
print("Session Insights (Semantica native):")
if isinstance(insights, dict):
    for k, v in insights.items():
        print(f"  {k}: {v}")
else:
    print(f"  {insights}")

In [ ]:

# Precedent search directly via Semantica's AgentContext
# (same data, no Agno in the loop)
precedents = store.context.find_precedents_advanced(
    scenario="borderline mortgage application",
    category="loan_approval",
)
print(f"\nPrecedent search via Semantica directly → {len(precedents or [])} results")

Summary

What	How
Persistent decision history	AgnoContextStore wrapping AgentContext + FAISS
Tool calls for decision intelligence	AgnoDecisionKit (record, find, trace, check, summarise)
Historical seeding	Native AgentContext.record_decision() — no Agno needed
Policy rules	Native PolicyEngine — no Agno needed
Post-session analytics	Native AgentContext.get_context_insights() — no Agno needed

The Agno integration is a thin wrapper — Semantica's full API remains directly accessible whenever you need finer control.