Agno × Semantica：决策智能智能体

Agno Agent
  ├── memory=AgnoContextStore       ← 图支撑的持久化记忆
  └── tools=[AgnoDecisionKit]       ← LLM 可调用的决策工具
          │
          ├── 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()

# 安装带 Agno 集成的 semantica
pip install semantica[agno]

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

# ── Semantica 核心（非 Agno 专用）──────────────────────────────────────
from semantica.context import AgentContext, ContextGraph
from semantica.context import PolicyEngine, DecisionQuery, CausalChainAnalyzer
from semantica.vector_store import VectorStore

# ── Agno 集成层 ───────────────────────────────────────────────────
from integrations.agno import AgnoContextStore, AgnoDecisionKit, AGNO_AVAILABLE

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

# ── 向量存储（FAISS，无需外部服务）────────────────────────
vector_store = VectorStore(backend="faiss", dimension=768)
print("VectorStore ready (FAISS)")

# ── 带完整分析能力的内存上下文图谱 ──────────────────────────────
knowledge_graph = ContextGraph(
    advanced_analytics=True,
    # 生产环境可切换为 neo4j：
    # backend="neo4j", uri="bolt://localhost:7687"
)
print("ContextGraph ready (in-memory)")

# 构建一个纯 Semantica 的 AgentContext 用于播种历史数据
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")

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

# 在接入 Agno 之前，先用 Semantica 的 PolicyEngine 直接验证
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)

# ── AgnoContextStore：将 AgentContext 包装为 Agno MemoryDb ────────────────────
store = AgnoContextStore(
    vector_store=vector_store,       # 同一个存储——共享已播种的决策
    knowledge_graph=knowledge_graph, # 同一个图谱——共享已播种的决策
    decision_tracking=True,
    graph_expansion=True,
    session_id="loan_underwriter_v1",
)
print("AgnoContextStore ready")

# ── AgnoDecisionKit：向 Agno 的 LLM 暴露 Semantica 决策工具 ──────────
decision_kit = AgnoDecisionKit(
    context=store.context,    # 复用同一个 AgentContext——共享决策历史
    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])

# 若 Agno 可用，则构建贷款审批智能体
if AGNO_AVAILABLE:
    from agno.agent import Agent
    from agno.memory import AgentMemory
    from agno.models.openai import OpenAIChat  # 或任何 Agno 支持的模型

    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")

import json

# ── 5a. 查找先例 ───────────────────────────────────────────────────────
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]}")

# ── 5b. 校验策略 ─────────────────────────────────────────────────────────
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', [])}")

# ── 5c. 记录决策 ──────────────────────────────────────────────────────
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']}")

# ── 5d. 分析影响 ───────────────────────────────────────────────────────
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}")

# ── 5e. 决策摘要 ─────────────────────────────────────────────────────
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}")

# 运行完整的 Agno 智能体：已安装 Agno 时由 LLM 自动编排决策工具，否则打印预期的推理流程
# 构造一个待审批的新贷款申请案例
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?"
)

# 已安装 Agno 时，由 LLM 自动编排 find_precedents / check_policy / record_decision
if AGNO_AVAILABLE:
    agent.print_response(NEW_CASE)
else:
    # 未安装 Agno 时，打印预期的智能体推理流程
    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")

# 直接从 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}")

# 直接通过 Semantica 的 AgentContext 进行先例检索
# （相同的数据，无需 Agno 参与）
precedents = store.context.find_precedents_advanced(
    scenario="borderline mortgage application",
    category="loan_approval",
)
print(f"\nPrecedent search via Semantica directly → {len(precedents or [])} results")