Semantica：给 AI Agent 装上「可审计的大脑」——开源版 Palantir 的图原生上下文基础设施

from semantica.context import ContextGraph, AgentContext
from semantica.vector_store import VectorStore

graph = ContextGraph(advanced_analytics=True)

# 带类型属性的节点
graph.add_node("acme_corp",    "Organization", name="Acme Corp", industry="SaaS")
graph.add_node("alice_chen",   "Person",       name="Alice Chen", role="CTO")
graph.add_node("contract_001", "Contract",     value=2_400_000, currency="USD")

# 带类型和元数据的边（额外 kwargs 自动成为边属性）
graph.add_edge("alice_chen", "acme_corp",    edge_type="works_for",  since="2019-03-01")
graph.add_edge("acme_corp",  "contract_001", edge_type="party_to",   signed="2024-01-15")

# BFS 遍历：从任意节点跳 N 跳
neighbors = graph.get_neighbors("acme_corp", hops=2)

# 时间点快照：图在过去任意日期的状态
snapshot  = graph.state_at("2024-01-01")

# AgentContext：面向 Agent 记忆工作流的高层 API
vs  = VectorStore(backend="faiss")
ctx = AgentContext(vector_store=vs, knowledge_graph=graph)
ctx.store("Alice approved the Acme renewal in Q1 2024", conversation_id="conv_001")
retrieved = ctx.retrieve("who approved the Acme contract?")

from semantica.context import ContextGraph

graph = ContextGraph(advanced_analytics=True)

# 记录带完整结构化上下文的决策
app_id = graph.record_decision(
    category="credit_application",
    scenario="Personal loan, $85k income, 31% DTI, 3yr employment",
    reasoning="Income meets threshold; employment stable; no adverse credit events",
    outcome="proceed_to_underwriting",
    confidence=0.88,
    metadata={"applicant_id": "A-7291"},
)
uw_id = graph.record_decision(
    category="loan_underwriting",
    scenario="Underwriting review for A-7291",
    reasoning="DTI within policy; clean 36-month credit history",
    outcome="approved",
    confidence=0.94,
)
rate_id = graph.record_decision(
    category="interest_rate",
    scenario="Rate assignment for approved loan A-7291",
    reasoning="Prime + 2.4% based on risk tier B2",
    outcome="rate_set_8.9pct",
    confidence=0.99,
)

# 构建可审计的因果链
# relationship_type 只能是 CAUSED / INFLUENCED / PRECEDENT_FOR 三者之一
graph.add_causal_relationship(app_id, uw_id,   relationship_type="CAUSED")
graph.add_causal_relationship(uw_id,  rate_id, relationship_type="INFLUENCED")

# 查询这套「智能」
chain     = graph.trace_decision_chain(rate_id)       # 完整因果祖先
similar   = graph.find_similar_decisions("personal loan approval, 31% DTI", max_results=5)
impact    = graph.analyze_decision_impact(uw_id)      # 下游影响面
compliant = graph.check_decision_rules({"category": "loan_underwriting", "confidence": 0.94})
insights  = graph.get_decision_insights()

from semantica.reasoning import ReteEngine, Rule, Fact, RuleType

rete = ReteEngine()
rete.build_network([
    Rule(
        rule_id="aml_flag",
        name="Flag high-risk transactions",
        conditions=[
            {"field": "amount",  "operator": ">",  "value": 10_000},
            {"field": "country", "operator": "in", "value": ["IR", "KP", "SY"]},
        ],
        conclusion="flag_for_compliance_review",
        rule_type=RuleType.IMPLICATION,
    ),
    Rule(
        rule_id="velocity_check",
        name="Flag rapid sequential transfers",
        conditions=[
            {"field": "transfers_in_1h", "operator": ">", "value": 5},
            {"field": "total_amount",    "operator": ">", "value": 50_000},
        ],
        conclusion="flag_velocity_breach",
        rule_type=RuleType.IMPLICATION,
    ),
])

rete.add_fact(Fact("tx_001", "transaction", [{"amount": 15_000, "country": "IR"}]))
flagged = rete.match_patterns()
# → [{"rule": "aml_flag", "matched_facts": ["tx_001"], "conclusion": "flag_for_compliance_review"}]

from semantica.reasoning import DatalogReasoner

engine = DatalogReasoner()
engine.add_fact("parent(tom, bob)")
engine.add_fact("parent(bob, ann)")
engine.add_fact("parent(ann, pat)")
engine.add_rule("ancestor(X, Y) :- parent(X, Y).")
engine.add_rule("ancestor(X, Z) :- parent(X, Y), ancestor(Y, Z).")
ancestors = engine.query("ancestor(tom, ?X)")
# → [{"X": "bob"}, {"X": "ann"}, {"X": "pat"}]

from semantica.reasoning import ExplanationGenerator, Reasoner

reasoner = Reasoner()
reasoner.add_fact("parent(tom, bob)")
reasoner.add_rule("ancestor(X, Y) :- parent(X, Y)")
result = reasoner.forward_chain()

explanation = ExplanationGenerator().generate_explanation(result)

from semantica.split import TextSplitter, RelationAwareChunker

text = open("contracts/master_agreement.txt").read()

# 实体感知：绝不把命名实体切开（GraphRAG 关键）
chunks = TextSplitter(method="entity_aware", ner_method="llm", chunk_size=1000).split(text)

# 关系感知：保持 (主语, 谓语, 宾语) 三元组完整
chunks = RelationAwareChunker(chunk_size=1000, preserve_triplets=True).chunk(text)

# 图切分：用中心性寻找自然的社区边界
chunks = TextSplitter(method="graph_based", chunk_size=1000).split(text)

# 层级切分：section → paragraph → sentence
chunks = TextSplitter(method="hierarchical", levels=["section", "paragraph"]).split(text)

from semantica.pipeline import PipelineBuilder, ExecutionEngine

builder = PipelineBuilder()

# 注意：add_step() 返回的是 PipelineStep 而非 builder，所以不能链式调用
builder.add_step("ingest",      step_type="ingest",           source="./contracts/", recursive=True)
builder.add_step("extract",     step_type="ner_extract")
builder.add_step("relations",   step_type="relation_extract")
builder.add_step("build_kg",    step_type="kg_build",         merge_entities=True)
builder.add_step("deduplicate", step_type="deduplicate",      threshold=0.75)
builder.add_step("export",      step_type="export",           format="turtle", output="kg.ttl")

# 而 connect_steps() 和 set_parallelism() 返回 builder，可以链式
pipeline = (
    builder
    .connect_steps("ingest",      "extract")
    .connect_steps("extract",     "relations")
    .connect_steps("relations",   "build_kg")
    .connect_steps("build_kg",    "deduplicate")
    .connect_steps("deduplicate", "export")
    .set_parallelism(4)
    .build(name="contracts_pipeline")
)

engine = ExecutionEngine()
result = engine.execute_pipeline(pipeline)

pytest tests/vector_store/test_performance_benchmarks.py -s

pip install semantica           # 核心
pip install semantica[all]      # 全家桶

pip install semantica[agno]                  # Agno 多 Agent 集成
pip install semantica[llm-litellm]           # OpenAI/Anthropic/Gemini/Mistral/Groq/Ollama/DeepSeek 等
pip install semantica[graph-neo4j]           # Neo4j (LPG)
pip install semantica[graph-falkordb]        # FalkorDB (LPG)
pip install semantica[graph-apache-age]      # Apache AGE (LPG)
pip install semantica[graph-amazon-neptune]  # AWS Neptune (LPG)
pip install semantica[tripletstore-oxigraph] # 内嵌 RDF 存储（内存/磁盘）
pip install semantica[vectorstore-qdrant]    # Qdrant
pip install semantica[vectorstore-pinecone]  # Pinecone
pip install semantica[db-snowflake]          # Snowflake
pip install semantica[db-databricks]         # Databricks
pip install semantica[ingest-parquet]        # Parquet / PyArrow
pip install semantica[ingest-arrow]          # Arrow / Feather / IPC
pip install semantica[viz]                   # HTML 交互式可视化
pip install semantica[explorer]              # Knowledge Explorer 仪表盘

semantica doctor
# Python 3.11.9         pass
# semantica 0.6.0       pass
# faiss vector store    pass
# Config file           pass    ~/.semantica/config.yaml

from semantica.context import ContextGraph

graph = ContextGraph(advanced_analytics=True)

# 每个 Agent 决策都变成可查询、可审计的知识节点
decision_id = graph.record_decision(
    category="vendor_selection",
    scenario="Choose cloud provider for HIPAA workload",
    reasoning="AWS offers BAA, mature HIPAA tooling, and existing team expertise",
    outcome="selected_aws",
    confidence=0.93,
)

# 问「为什么会这样」，得到真正结构化的答案
chain     = graph.trace_decision_chain(decision_id)
similar   = graph.find_similar_decisions("cloud vendor", max_results=5)
impact    = graph.analyze_decision_impact(decision_id)
compliant = graph.check_decision_rules({"category": "vendor_selection"})

git clone https://github.com/semantica-agi/semantica.git
cd semantica && pip install -e ".[dev]" && pytest tests/

FROM node:26-alpine AS frontend-builder
WORKDIR /app/explorer
RUN npm ci && npm run build

FROM python:3.14-slim AS runtime
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 \
    FALKORDB_HOST=falkordb FALKORDB_PORT=6379

RUN groupadd --system semantica \
    && useradd --system --gid semantica --home-dir /app --shell /usr/sbin/nologin semantica

RUN pip install --no-cache-dir ".[explorer]" && chown -R semantica:semantica /app
USER semantica
EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 \
    CMD python -c "import json, urllib.request; data=json.load(urllib.request.urlopen('http://127.0.0.1:8000/api/health', timeout=3)); raise SystemExit(0 if data.get('status') == 'ok' else 1)"

CMD ["python", "-m", "uvicorn", "semantica.explorer.app:app", "--host", "0.0.0.0", "--port", "8000"]

from semantica.ingest import FileIngestor, WebIngestor, ParquetIngestor, DBIngestor

# 整目录合同（PDF/DOCX/HTML/TXT）
docs = FileIngestor().ingest_directory("./contracts/", recursive=True)

# 遵守 robots.txt 的 Web 摄取
pages = WebIngestor().ingest_url("https://example.com/reports/annual-2024.html")

# Parquet（Snappy 压缩）
records = ParquetIngestor().ingest("./data/transactions.parquet")

# SQL 数据库，指定拉哪些表
rows = DBIngestor().ingest_database(
    connection_string="postgresql://user:pass@localhost/mydb",
    include_tables=["customer_events"],
    max_rows_per_table=50_000,
)

from semantica.ingest import DatabricksIngestor, SnowflakeIngestor

databricks = DatabricksIngestor(
    host="https://adb-xxx.azuredatabricks.net",
    token="dapi-xxxxxxxx",              # 或用 client_id/client_secret 走 OAuth M2M
    http_path="/sql/1.0/warehouses/xxxxxxxx",
    catalog="main",
)
customers     = databricks.ingest_table("customers", limit=10_000)
sales         = databricks.ingest_query("SELECT * FROM sales WHERE region = 'EMEA'")
table_lineage = databricks.get_table_lineage("customers", catalog="main", schema="default")

snowflake = SnowflakeIngestor(
    account="myaccount", user="myuser", password="mypassword",
    warehouse="COMPUTE_WH", database="MYDB",
)
orders = snowflake.ingest_table("ORDERS", limit=10_000)

from semantica.semantic_extract import (
    NamedEntityRecognizer, RelationExtractor, EventDetector, TripletExtractor,
)

text = """
Anthropic CEO Dario Amodei announced a $7.3B Series E funding round in partnership
with Google and Spark Capital, valuing the company at $61.5B as of Q4 2024.
"""

ner = NamedEntityRecognizer(confidence_threshold=0.7)
entities = ner.extract_entities(text)
# → [Entity(name="Dario Amodei", type="PERSON"), Entity(name="Anthropic", type="ORG"), ...]

relations = RelationExtractor(confidence_threshold=0.6, bidirectional=True) \
    .extract_relations(text, entities=entities)
# → [Relation(subject="Dario Amodei", predicate="ceo_of", object="Anthropic"), ...]

events = EventDetector(extract_participants=True, extract_time=True).detect_events(text)
# → [Event(type="FUNDING", participants=["Anthropic","Google","Spark Capital"], ...)]

triplets = TripletExtractor(include_temporal=True, include_provenance=True).extract_triplets(text)

from semantica.ingest import FileIngestor
from semantica.kg import (
    GraphBuilder, GraphAnalyzer, CentralityCalculator,
    CommunityDetector, PathFinder, LinkPredictor, BiTemporalFact,
)
from datetime import datetime

sources = FileIngestor().ingest_directory("./contracts/", recursive=True)
kg = GraphBuilder(merge_entities=True, enable_temporal=True).build(sources)

analysis    = GraphAnalyzer().analyze_graph(kg)
degree      = CentralityCalculator().calculate_degree_centrality(kg)   # 最核心的实体
communities = CommunityDetector().detect_communities(kg, method="louvain")
path        = PathFinder().find_shortest_path(kg, "alice_chen", "contract_001")
predictions = LinkPredictor().predict_links(kg, top_k=10)

# 双时态：有效时间（世界上何时为真）与记录时间（你何时知道）互相独立
fact = BiTemporalFact(
    valid_from=datetime(2024, 3, 1),
    valid_until=datetime(2025, 1, 1),
    recorded_at=datetime(2024, 3, 5),
)

from semantica.context import ContextGraph
from semantica.provenance import ProvenanceManager
from semantica.export import RDFExporter

graph = ContextGraph(advanced_analytics=True)
prov  = ProvenanceManager(storage_path="./audit.db")

# 1. 记录决策链
d1 = graph.record_decision(
    category="drug_interaction_check",
    scenario="Patient P-4821: warfarin + amiodarone co-prescribed",
    reasoning="Amiodarone potentiates warfarin's anticoagulant effect",
    outcome="flag_for_review", confidence=0.91,
)
d2 = graph.record_decision(
    category="dosage_adjustment",
    scenario="INR monitoring plan for P-4821",
    reasoning="Reduce warfarin dose per interaction severity; recheck INR in 5 days",
    outcome="dose_reduced_30pct", confidence=0.87,
)
graph.add_causal_relationship(d1, d2, relationship_type="CAUSED")

# 2. 为每个实体挂溯源
prov.track_entity(
    "patient_P4821",
    source="ehr/medication_orders_2024.json",
    metadata={"extractor": "NamedEntityRecognizer"},
)

# 3. 导出 W3C PROV-O 供监管提交
#    注意：RDFExporter 期望 {"entities": [...], "relationships": [...]}，
#    而 ContextGraph.to_dict() 给的是 {"nodes": [...], "edges": [...]}，需要先映射
graph_dict = graph.to_dict()
kg = {
    "entities": [
        {"id": n["id"], "type": n["type"], "text": n["content"]}
        for n in graph_dict["nodes"]
    ],
    "relationships": [
        {"source_id": e["source"], "target_id": e["target"], "type": e["type"]}
        for e in graph_dict["edges"]
    ],
}
RDFExporter().export(kg, "audit_trail.ttl", format="turtle")

from semantica.vector_store import VectorStore, HybridSearch

# inmemory 后端下 HybridSearch 与 explain_decision() 都开箱可用
# 规模上来后换成 qdrant / weaviate / milvus / pinecone / pgvector / faiss，API 完全一致
vs = VectorStore(backend="inmemory", dimension=1536)

vs.store_decision(
    scenario="Personal loan A-7291, $85k income, 31% DTI, 3yr employment",
    outcome="approved", confidence=0.94, category="loan_underwriting",
)

results = vs.search(query="personal loan approval with low DTI", limit=10)

# 稠密 + 稀疏一次检索，RRF 融合
hits = HybridSearch(vector_store=vs).search("high-risk transactions 2024")

# 解释某条决策为什么被检索出来
explanation = vs.explain_decision(results[0]["id"])

semantica         # 启动仪表盘
semantica doctor  # 健康检查
semantica --help  # 完整分组命令参考

# ❌ 这样会 ImportError
# from semantica.ingest import DuckDBIngestor

# ✅ 正确写法
from semantica.ingest.duckdb_ingestor import DuckDBIngestor

pytest tests/vector_store/test_performance_benchmarks.py -s