Agno × Semantica：GraphRAG 上下文智能体

Agno Agent
  ├── knowledge=AgnoKnowledgeGraph       ← GraphRAG 知识库
  └── tools=[AgnoKGToolkit]              ← 实时的图谱构建/查询工具
          │
          │   由 Semantica 支撑：
          ├── NERExtractor               ← 命名实体识别
          ├── RelationExtractor          ← 关系抽取
          ├── GraphBuilder               ← 从抽取结果构建 ContextGraph
          ├── ContextGraph               ← 带分析能力的内存图谱
          └── Reasoner                   ← 基于规则的推断

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

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

# ── Semantica 核心——直接用于流水线设置 ───────────────────────
from semantica.kg import GraphBuilder
from semantica.context import ContextGraph
from semantica.semantic_extract import NERExtractor, RelationExtractor, TripletExtractor
from semantica.reasoning import Reasoner
from semantica.vector_store import VectorStore

# ── Agno 集成层 ───────────────────────────────────────────────────
from integrations.agno import AgnoKnowledgeGraph, AgnoKGToolkit, AGNO_AVAILABLE

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

# NER——识别组织、法规、日期、金额、角色
ner = NERExtractor()

# 关系抽取器——发现实体之间的带类型边
rel_extractor = RelationExtractor(confidence_threshold=0.60)

# 知识图谱构建器
graph_builder = GraphBuilder(merge_entities=True, temporal_support=True)

# 内存上下文图谱（如需持久化可切换为 neo4j/falkordb）
context_graph = ContextGraph(advanced_analytics=True)

# 用于在图谱上进行规则推断的推理器
reasoner = Reasoner()

print("Semantica extraction pipeline assembled")

# 监管文档（代表性片段）
REGULATORY_DOCS = [
    {
        "title": "Basel IV — Capital Requirements",
        "text": (
            "Basel IV introduces a revised standardised approach for credit risk, "
            "replacing internal model floors. Banks must maintain a minimum CET1 ratio "
            "of 4.5% and a total capital ratio of 8%. The BCBS finalised these requirements "
            "in December 2017 with a phased implementation starting January 2022. "
            "National regulators including the EBA and FCA are responsible for local "
            "transposition. Risk-weighted assets under Basel IV are calculated using "
            "the Output Floor, capping RWA reductions at 72.5%."
        ),
    },
    {
        "title": "DORA — Digital Operational Resilience Act",
        "text": (
            "DORA (Regulation EU 2022/2554) applies to financial entities and ICT "
            "third-party service providers operating in the EU. It mandates ICT risk "
            "management frameworks, incident classification, and annual operational "
            "resilience testing. Supervised entities must report major ICT incidents to "
            "the European Supervisory Authorities (ESAs) within 4 hours of classification. "
            "Critical ICT providers are subject to direct oversight by the Joint Oversight "
            "Network led by ESMA, EBA, and EIOPA. DORA became applicable on 17 January 2025."
        ),
    },
    {
        "title": "AML — Anti-Money Laundering Directive VI",
        "text": (
            "AMLD6 strengthens the EU's anti-money laundering framework by extending "
            "criminal liability to 22 predicate offences including cybercrime and "
            "environmental crime. Financial institutions must apply Customer Due Diligence "
            "(CDD) at onboarding and Enhanced Due Diligence (EDD) for high-risk customers. "
            "Suspicious Activity Reports (SARs) are filed with the national Financial "
            "Intelligence Unit (FIU). Non-compliance carries penalties up to 10% of "
            "annual global turnover. AMLD6 was transposed into UK law via MLCO 2020."
        ),
    },
]

print(f"Documents to ingest: {len(REGULATORY_DOCS)}")
for doc in REGULATORY_DOCS:
    print(f"  • {doc['title']}")

# ── 直接使用 Semantica 运行 NER ─────────────────────────────────────────
all_entities = []
for doc in REGULATORY_DOCS:
    entities = ner.extract_entities(doc['text']) or []
    all_entities.extend(entities)
    print(f"[{doc['title']}] → {len(entities)} entities")
    for e in entities[:4]:
        print(f"    {getattr(e,'name','?'):30s}  type={getattr(e,'type','?')}  conf={getattr(e,'confidence',0):.2f}")

print(f"\nTotal entities extracted: {len(all_entities)}")

# ── 直接使用 Semantica 运行关系抽取 ──────────────────────────
all_relations = []
for doc in REGULATORY_DOCS:
    relations = rel_extractor.extract_relations(doc['text']) or []
    all_relations.extend(relations)
    print(f"[{doc['title']}] → {len(relations)} relations")
    for r in relations[:3]:
        src  = getattr(r, 'source', '?')
        rtype = getattr(r, 'type', getattr(r, 'relation', '?'))
        tgt  = getattr(r, 'target', '?')
        conf = getattr(r, 'confidence', 0)
        print(f"    {src!s:20s} --[{rtype}]--> {tgt!s:20s}  conf={conf:.2f}")

print(f"\nTotal relations extracted: {len(all_relations)}")

kg = AgnoKnowledgeGraph(
    graph_builder=graph_builder,
    ner_extractor=ner,
    relation_extractor=rel_extractor,
    context_graph=context_graph,
    num_documents=5,
)

# 通过集成封装摄取所有文档
kg.load(texts=[doc['text'] for doc in REGULATORY_DOCS])

print(f"AgnoKnowledgeGraph: {len(kg._docs)} documents indexed")

# 用一组监管合规问题测试 GraphRAG 的多跳检索能力
queries = [
    "What is the minimum CET1 ratio required under Basel IV?",
    "Which authorities supervise critical ICT providers under DORA?",
    "What are the reporting timelines for major ICT incidents?",
    "How does AMLD6 handle customer due diligence?",
]

# 对每个问题执行多跳图检索，并打印命中的文档片段
for query in queries:
    print(f"\nQ: {query}")
    results = kg.search(query, num_documents=2)
    print(f"   Retrieved {len(results)} document(s)")
    for i, doc in enumerate(results, 1):
        content = getattr(doc, 'content', str(doc))
        print(f"   [{i}] {content[:120]}...")

# 获取特定实体的图谱上下文
entity_contexts = ["BCBS", "EBA", "DORA", "Basel IV"]
for entity in entity_contexts:
    ctx = kg.get_graph_context(entity)
    print(f"\nGraph context for '{entity}':")
    print(ctx if ctx else "  (no graph nodes found — depends on NER extraction quality)")

# 构建 Agno 知识图谱工具包，供智能体调用
toolkit = AgnoKGToolkit(
    ner_extractor=ner,
    relation_extractor=rel_extractor,
    reasoner=reasoner,
    context=context_graph,  # 与知识库共享同一个图谱
)

print(f"AgnoKGToolkit: {len(toolkit._tools)} tools")
print("  Tools:", [fn.__name__ for fn in toolkit._tools])

# 工具：extract_entities
print("=" * 55)
print("TOOL: extract_entities")
print("=" * 55)

new_text = (
    "The PRA published a consultation paper requiring UK banks to "
    "implement DORA-equivalent resilience testing by Q3 2025, "
    "with Barclays and HSBC named as systemic institutions."
)
entities_json = toolkit.extract_entities(new_text)
entities_result = json.loads(entities_json)
print(f"Found {entities_result['count']} entities:")
for e in entities_result['entities']:
    print(f"  {e['name']:30s}  type={e['type']:15s}  conf={e['confidence']:.2f}")

# 工具：extract_relations
print("=" * 55)
print("TOOL: extract_relations")
print("=" * 55)

relations_json = toolkit.extract_relations(new_text)
relations_result = json.loads(relations_json)
print(f"Found {relations_result['count']} relations:")
for r in relations_result['relations']:
    print(f"  {r['source']:20s} --[{r['relation']}]--> {r['target']:20s}")

# 工具：add_to_graph
print("=" * 55)
print("TOOL: add_to_graph")
print("=" * 55)

add_result = json.loads(toolkit.add_to_graph(
    entities=json.dumps([
        {"name": "PRA", "type": "REGULATOR"},
        {"name": "Barclays", "type": "BANK"},
        {"name": "HSBC", "type": "BANK"},
    ]),
    relations=json.dumps([
        {"source": "PRA", "relation": "SUPERVISES", "target": "Barclays"},
        {"source": "PRA", "relation": "SUPERVISES", "target": "HSBC"},
        {"source": "Barclays", "relation": "SUBJECT_TO", "target": "DORA"},
        {"source": "HSBC", "relation": "SUBJECT_TO", "target": "DORA"},
    ]),
))
print(f"Added: {add_result['nodes_added']} nodes, {add_result['edges_added']} edges")

# 工具：query_graph
print("=" * 55)
print("TOOL: query_graph")
print("=" * 55)

query_result = json.loads(toolkit.query_graph("PRA"))
print(f"Keyword query 'PRA' → {query_result['count']} node(s):")
for node in query_result['results']:
    print(f"  label={node.get('label')}  type={node.get('type')}")

# 工具：find_related
print("=" * 55)
print("TOOL: find_related")
print("=" * 55)

related_result = json.loads(toolkit.find_related("Barclays", hops=2))
print(f"Related to 'Barclays' (2 hops): {related_result['count']} entity/entities")
for name in related_result['related']:
    print(f"  → {name}")

# 工具：infer_facts——Semantica 的 Reasoner 从图谱状态推导出新事实
print("=" * 55)
print("TOOL: infer_facts")
print("=" * 55)

# 规则：监管合规推断
inference_rules = json.dumps([
    "IF BANK(?x) THEN FinancialEntity(?x)",
    "IF REGULATOR(?x) THEN SupervisoryAuthority(?x)",
    "IF FinancialEntity(?x) THEN ComplianceSubject(?x)",
])

infer_result = json.loads(toolkit.infer_facts(rules=inference_rules))
print(f"Inferred {infer_result['count']} new fact(s):")
for fact in infer_result['inferred_facts'][:8]:
    print(f"  {fact}")

# 工具：export_subgraph——为下游系统导出知识
print("=" * 55)
print("TOOL: export_subgraph (JSON-LD)")
print("=" * 55)

export_result = json.loads(toolkit.export_subgraph(entity="DORA", format="json-ld"))
print(f"Exported as format='{export_result['format']}'")
if 'data' in export_result:
    preview = str(export_result['data'])[:300]
    print(f"Preview: {preview}...")
elif 'nodes' in export_result:
    print(f"Graph nodes exported: {len(export_result['nodes'])}")
    for node in export_result['nodes'][:5]:
        print(f"  {node}")

# 运行完整的 Agno GraphRAG 智能体：已安装 Agno 时由 LLM 结合知识图谱回答合规问题
if AGNO_AVAILABLE:
    from agno.agent import Agent
    from agno.models.openai import OpenAIChat

    # 构建合规分析智能体，挂载 GraphRAG 知识库与实时图谱工具
    compliance_agent = Agent(
        name="ComplianceAnalyst",
        model=OpenAIChat(id="gpt-4o"),
        knowledge=kg,
        search_knowledge=True,
        tools=[toolkit],
        show_tool_calls=True,
        description=(
            "You are a regulatory compliance analyst. Use the knowledge graph "
            "to answer questions about Basel IV, DORA, and AML regulations. "
            "When answering, use find_related and query_graph to discover "
            "connections between regulators, rules, and institutions."
        ),
    )

    # 向智能体提出跨 DORA 与 Basel IV 的合规问题
    compliance_agent.print_response(
        "Which supervisory authorities are responsible for overseeing DORA compliance "
        "for UK banks, and how does this relate to Basel IV capital requirements?"
    )
else:
    # 未安装 Agno 时，打印预期的多跳推理流程
    print("[Agno not installed — skipping live agent run]")
    print()
    print("Expected reasoning flow:")
    print("  search_knowledge('DORA supervisory authorities UK banks')")
    print("    → retrieves DORA doc with graph expansion")
    print("  query_graph('PRA') → finds PRA node")
    print("  find_related('PRA', hops=2) → PRA → SUPERVISES → Barclays, HSBC")
    print("  find_related('Basel IV', hops=1) → capital ratio requirements")
    print("  Answer: PRA supervises UK banks under DORA; Basel IV CET1 requirement is 4.5%")

# 在同一个 ContextGraph 上直接使用 Semantica 的 GraphAnalyzer
from semantica.kg import GraphAnalyzer, CentralityCalculator, PathFinder

try:
    analyzer = GraphAnalyzer()
    analysis = analyzer.analyze_graph(context_graph)
    print("Graph analysis (Semantica native):")
    if isinstance(analysis, dict):
        for k, v in list(analysis.items())[:8]:
            print(f"  {k}: {v}")
    else:
        print(f"  {analysis}")
except Exception as e:
    print(f"GraphAnalyzer: {e}")

# 中心性——哪些实体连接最多/最具影响力？
try:
    centrality = CentralityCalculator()
    scores = centrality.calculate_degree_centrality(context_graph)
    print("Degree centrality (most connected entities):")
    if isinstance(scores, dict):
        top = sorted(scores.items(), key=lambda x: x[1], reverse=True)[:5]
        for entity, score in top:
            print(f"  {entity:30s} {score:.4f}")
    else:
        print(f"  {scores}")
except Exception as e:
    print(f"CentralityCalculator: {e}")