基于上下文和可追溯 AI 系统的图原生基础设施

摄入你的企业数据，提取关键内容，构建上下文图谱和知识图谱（KG），并在其上运行图分析和因果推理，同时内置完整的决策溯源。可解释、可追溯、可信赖，从设计层面实现。

决策智能 · 上下文管理 · 确定性推理 · 本体管理 · 知识建模 · 端到端可追溯性

开源 · 可自托管 · 可审计 · 可治理 · 零供应商锁定

多语言图存储 · RDF 和 LPG 支持 · W3C 标准 · 互操作性强

pip install semantica

知识浏览器 · 上下文图谱 · 推理引擎 · 决策智能 · 本体中心 · 治理与本体 · 审计追踪 · 决策智能 · 上下文图谱 · 配方：审计追踪 · 模块参考 · 集成 · CLI · 性能 · 安装

▶ 观看完整平台演示

大多数 AI 智能体行事无迹可寻。它们存储的是向量嵌入，而非真正的语义：无法解释的上下文，无法审计的决策。在信贷领域，这个缺口不是麻烦，而是合规风险——一个承保智能体的审批决定，必须在数月后经受监管机构的「为什么」追问。

Semantica 位于你的 LLM、向量存储和智能体框架之下，作为确定性基础设施层：图构建、推理和溯源均无需 LLM。

目标用户

构建 AI/ML 平台的团队，交付需要基于碎片化原始数据构建结构化、可查询上下文（而非仅向量索引）的智能体，并做出重大决策

使用 Databricks 或 Snowflake 的数据平台团队，希望将已存在于 Unity Catalog 或 Snowflake 数据仓库中的表直接转化为受治理、有血缘追踪的知识图谱，而无需先将数据导出到第三方 SaaS

合规、风控和审计团队，需要以监管机构实际接受的方式获得「AI 为什么这样做」的明确答案

受监管企业（金融、医疗、法律、政府、国防），不能交付黑箱产品，也不能将数据发送到别人的 SaaS 来获得答案

平台和基础设施工程师，希望知识图谱、推理和溯源技术栈可自托管、可替换，不被锁定在某一供应商的后端

数据和知识工程师，从混乱、多源数据构建知识图谱：实体和关系被提取，冲突或矛盾的事实被标记而非静默覆盖，重复项在变成噪声之前被合并

快速入门 · 架构 · 你将获得什么 · 为什么选择 Semantica · 决策智能 · 上下文图谱 · 配方：审计追踪 · 模块参考 · 集成 · CLI · 性能 · 安装

Semantica 为你提供的能力

上下文图谱：结构化、可查询的图谱，记录你的智能体知道、决定和推理的一切

决策智能：每个决策都是一等公民：可追溯、按先例可搜索，并因果关联

AI 治理与本体：SHACL 约束、冲突检测、合规规则、OWL 生成，以及 SKOS 词汇管理，配合可视化编辑器

完整可审计性：每个事实都遵循 W3C PROV-O 溯源，审计日志可导出为 JSON、CSV 或 RDF

确定性推理：前向链式推理、Rete 网络、Datalog 和 SPARQL，路径完全可解释，不是黑箱

知识流水线：多源摄入、实体感知分块、NER/关系/事件提取和知识图谱构建，全程保留语义的去重和溯源合并

企业数据平台：Databricks（Unity Catalog + Delta Lake，PAT/OAuth M2M 认证，目录/模式/表/血缘内省）和 Snowflake（数据仓库/数据库/模式，密钥对和 OAuth 认证）的原生连接器，让已存在于数据湖或数据仓库中的表成为带溯源的图节点，而非又一次导出/导入的跳板

图分析：在你刚构建的图上运行中心性、社区检测、链接预测和最短路径查询

多语言图存储：原生 RDF（嵌入式 Oxigraph、Blazegraph、Apache Jena、Eclipse RDF4J，通过 SPARQL）和标签属性图（Neo4j、FalkorDB、Apache AGE、AWS Neptune，通过 Cypher），以及向量存储，全部可替换而不影响你的代码

可视化：在交互式浏览器工作台中探索任意图谱、本体或时间线

即插即用集成：原生 Agno 支持、功能完整的 MCP 服务器、全面的 CLI、REST API，以及覆盖主流编辑器的插件

Semantica 对现有技术栈进行补充，而非替换它。保留你的 LLM、向量数据库和智能体框架一切照旧；Semantica 在其之上叠加决策记录、因果推理溯源、本体治理、冲突检测和审计追踪。推理引擎、知识图谱构建和溯源层完全确定性；使用它们无需调用 LLM。

pip install semantica

from semantica.context import ContextGraph

graph = ContextGraph(advanced_analytics=True)

# 每个智能体决策都成为一个可查询、可审计的知识节点
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"})  # 策略门控

5 秒验证安装：

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

如果 Semantica 解决了你的实际问题，一颗 star 能帮助更多人找到它。

⭐ Star on GitHub · Join Discord

Semantica 是一个真正的端到端流水线，而非一个挂着营销名的单一库。以下每个阶段都是一个可发货的模块，可独立导入：

Sources → Ingest → Parse → Normalize → Split → Extract → Conflict Detection → Deduplication
   → Knowledge Graph → [ Ontology · Reasoning · Provenance · Decisions ] → Enriched KG
   → Vector Store + Polyglot Graph Store (RDF & LPG) → Export / Visualize / REST · MCP · CLI

Ingest（摄取）：文件、Web、数据库、企业数据平台（Databricks、Snowflake）、云（Google Drive、Elasticsearch）、流（Kafka、Kinesis）、Git、邮件、MCP

Parse → Normalize → Split：文档解析、文本/实体/日期规范化、GraphRAG 原生的实体感知分块

Extract → Conflict Detection → Deduplication：NER、关系、事件、三元组；冲突事实在合并前被标记并解决

Knowledge Graph：GraphBuilder 构建图谱；双时态事实和完整图分析（中心性、社群、链接预测）运行于其上

Ontology · Reasoning · Provenance · Decisions：位于 KG 之上的智能层，包含 SHACL/OWL 治理、Rete/Datalog/SPARQL 推理、W3C PROV-O 血缘追踪，以及一等公民级别的决策记录

Storage：多语言存储设计，包含 RDF 三元组存储（嵌入式 Oxigraph、Blazegraph、Apache Jena、Eclipse RDF4J）、标属性图（Neo4j、FalkorDB、Apache AGE、AWS Neptune）和向量数据库，全部可无代码切换

Outputs：导出（RDF、OWL、Parquet、Cypher、JSON-LD）、交互式可视化，以及通过 REST API、MCP 服务器或 CLI 访问

→ 流水线和决策智能生命周期的完整 Mermaid 图表

决策智能

决策智能将每个 AI 选择从临时推理转化为永久、可审计、可查询的记录。它回答"你的 AI 做了什么决定、为什么、以及接下来发生了什么？"：这是监管机构和 enterprise 风险团队日益迫切地问出的问题。

在 Semantica 中，决策不是一行日志。它是一个具有完整生命周期的顶级图节点。在受监管领域，每个 AI 决策都必须能追溯到来源并能向审计员辩护：record_decision() 创建一个永久的、结构化的记录，可导出为 W3C PROV-O——这是大多数合规框架接受的监管提交格式。

record_decision()             → 作为具有完整结构化上下文的图节点存储
add_causal_relationship()     → 链接到上游原因和下游影响
find_similar_decisions()      → 跨所有历史决策的语义先例搜索
trace_decision_chain()        → 追溯到根本原因的完整因果血统
analyze_decision_impact()     → 下游影响图——此决策影响的所有内容
check_decision_rules()        → 针对可配置规则集的策略合规门控
export / audit trail          → W3C PROV-O、CSV 或 JSON，用于监管提交

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",
    outcome="rate_set_8.9pct",
    reasoning="Prime + 2.4% based on risk tier B2",
    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()

上下文图谱是传统 RAG 缺失的结构化记忆层。不同于回答"什么是相似的？"的平面嵌入，上下文图谱回答"什么是连接的、为什么连接、以及如何连接的？"。每个实体、关系、决策和事实都是顶级节点，可通过图遍历进行查询。实体链接到源文档，决策链接到证据和后果，事实携带完整溯源，冲突被检测而非被静默覆盖。

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 遍历 - 从任意节点出发跳步穿越图谱
neighbors = graph.get_neighbors("acme_corp", hops=2)

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

# AgentContext - 智能体记忆工作流的高级 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?")

为什么用图谱而非嵌入：遍历能发现嵌入会遗漏的连接（一个与合同相距 3 跳的人）；每个节点都携带溯源信息，因此你始终可以问"这从哪里来的？"；冲突在被合并前就被标记，防止它们污染你的知识库；时间点快照让你无需重新处理即可回放历史。

配方：受监管决策的审计追踪

旗舰模式：记录因果关联的决策链，为每个实体附加溯源，并导出发送给监管者的审计追踪。

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

# Record the decision chain

```python
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,
)
# relationship_type must be one of CAUSED, INFLUENCED, or PRECEDENT_FOR
graph.add_causal_relationship(d1, d2, relationship_type="CAUSED")

Track provenance for every entity

prov.track_entity("patient_P4821", source="ehr/medication_orders_2024.json",
                  metadata={"extractor": "NamedEntityRecognizer"})

Export W3C PROV-O for regulator submission

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

更多示例（GraphRAG 管道、AML 规则引擎、单次 ontology-to-KG 转换）见下方「更多示例」章节。

每个模块均可独立导入，且均附有经当前源码树验证过的可运行代码示例；可单独使用，也可组合使用。

↓ 点击下方模块参考展开各模块的可运行示例。

支持从文件、Web、数据库、API、消息流、邮件、Git 仓库、Parquet、Databricks、Snowflake 或 MCP 服务器摄取数据，均通过统一接口实现。

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

# Ingest an entire directory of contracts (PDF, DOCX, HTML, TXT)
docs = FileIngestor().ingest_directory("./contracts/", recursive=True)

# Ingest live web content with robots.txt compliance
pages = WebIngestor().ingest_url("https://example.com/reports/annual-2024.html")

# Ingest structured data from Parquet with Snappy compression
records = ParquetIngestor().ingest("./data/transactions.parquet")

# Ingest from a SQL database - specify which tables to pull
rows = DBIngestor().ingest_database(
    connection_string="postgresql://user:pass@localhost/mydb",
    include_tables=["customer_events"],
    max_rows_per_table=50_000,
)

# Enterprise data platforms - pull tables straight out of your lakehouse
# or warehouse, with lineage, instead of exporting to CSV first
from semantica.ingest import DatabricksIngestor, SnowflakeIngestor

# pip install "semantica[db-databricks]"
databricks = DatabricksIngestor(
    host="https://adb-xxx.azuredatabricks.net",
    token="dapi-xxxxxxxx",              # or client_id/client_secret for 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")  # Unity Catalog lineage

# pip install semantica[db-snowflake]
snowflake = SnowflakeIngestor(
    account="myaccount",
    user="myuser",
    password="mypassword",              # or private_key=... for key-pair; use authenticator="oauth", token=... for OAuth
    warehouse="COMPUTE_WH",
    database="MYDB",
)
orders = snowflake.ingest_table("ORDERS", limit=10_000)

Security Note: 生产环境中切勿将凭据（token、password、private_key）硬编码；请通过环境变量（如 DATABRICKS_TOKEN、SNOWFLAKE_PASSWORD）或密钥管理器传递。

支持的数据源：本地文件（PDF、DOCX、PPTX、HTML、TXT、CSV、JSON、YAML、Excel、XML）· 网页· RSS/Atom 订阅源· REST API· 数据库（PostgreSQL、MySQL、SQLite、Oracle、SQL Server）· Parquet 数据集· Databricks（Unity Catalog + Delta Lake）· Snowflake· Git 仓库· 邮件（IMAP/POP3）· 消息流（Kafka、RabbitMQ、Kinesis、Pulsar）· MCP 资源· Apache Arrow/Feather/IPC（ArrowIngestor）

DuckDB、Elasticsearch、Google Drive、HuggingFace、MongoDB 和 Pandas 的摄取功能也已打包（DuckDBIngestor、ElasticIngestor、GDriveIngestor、HuggingFaceIngestor、MongoIngestor、PandasIngestor），但尚未从顶层 semantica.ingest 命名空间重新导出——需直接导入：from semantica.ingest.duckdb_ingestor import DuckDBIngestor。

从原始文本中一次性提取结构化知识。

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

# Named entity recognition with confidence thresholding
ner = NamedEntityRecognizer(confidence_threshold=0.7)
entities = ner.extract_entities(text)
# → [Entity(name="Dario Amodei", type="PERSON"), Entity(name="Anthropic", type="ORG"),
#    Entity(name="Google", type="ORG"), Entity(name="$7.3B", type="MONEY"), ...]

# Relationship extraction - bidirectional support
rel_extractor = RelationExtractor(confidence_threshold=0.6, bidirectional=True)
relations = rel_extractor.extract_relations(text, entities=entities)
# → [Relation(subject="Dario Amodei", predicate="ceo_of", object="Anthropic"),
#    Relation(subject="Anthropic", predicate="raised", object="$7.3B Series E"), ...]

# Event detection with temporal processing
events = EventDetector(extract_participants=True, extract_time=True).detect_events(text)
# → [Event(type="FUNDING", participants=["Anthropic","Google","Spark Capital"],
#          amount="$7.3B", date="Q4 2024")]

# RDF triplets with optional provenance metadata
triplets = TripletExtractor(include_temporal=True, include_provenance=True).extract_triplets(text)
# → [("Anthropic", "valuation", "$61.5B"), ("Dario Amodei", "is_ceo_of", "Anthropic"), ...]

跨多文档的批处理使用 ner.process_batch([...])，而非门面类上的逐调用 extract_entities_batch。

从文档构建生产级知识图谱，并对其运行图算法。

from semantica.graph import ContextGraph, GraphAlgorithms

graph = ContextGraph()

# Build from structured records
graph.from_records(records, entity_types=["PERSON", "ORG", "MONEY"])

# Find shortest path between two entities
path = GraphAlgorithms.shortest_path(graph, "Anthropic", "Google")

# Detect communities (useful for corpus-level topic clustering)
communities = GraphAlgorithms.louvain_communities(graph)

# Node importance via PageRank
ranks = GraphAlgorithms.page_rank(graph)