Ontology 最小 MVP：和普通业务系统的区别，以及一步一步怎么做

---
title: Ontology 最小 MVP：和普通业务系统的区别，以及一步一步怎么做
published_at: 2026-08-05
language: zh
---

# Ontology 最小 MVP：和普通业务系统的区别，以及一步一步怎么做

做 Ontology MVP，不能一上来就说“建几个对象、连几个关系、做个风险判断”。那样听起来和正常开发一个业务系统没有区别，也确实没有解释清楚 Ontology 的价值。

正确顺序应该是：

```text
先讲清楚它和普通业务系统有什么区别；
再讲清楚本体层的优势在哪里；
再讲清楚 AI 为什么需要这层本体；
最后用一个具体场景，一步一步搭出最小 MVP。
```

本文用“订单延期风险分析”作为例子，分别讲三条路线：

```text
路线 A：使用 Palantir 官方平台搭 MVP；
路线 B：仍使用 Palantir，但不用官方 OSDK，直接调 Ontology REST API；
路线 C：完全不用 Palantir，自建一个轻量 Ontology Runtime。
```

目标不是泛泛讲架构，而是让这个 MVP 能真的一步一步搭起来。

## 一、普通业务系统和 Ontology 系统有什么区别

如果只做一个订单延期风险系统，普通开发方式通常是：

```text
建数据库表；
写后端接口；
写前端页面；
写风险规则；
加任务创建按钮；
接一个 AI 解释接口。
```

这没有问题，但它本质上还是一个普通业务系统。

普通业务系统的特点是：

```text
业务含义藏在代码里；
对象关系藏在 SQL join 里；
动作藏在前端按钮和后端接口里；
权限分散在不同接口里；
AI 需要额外手写 tool 和 prompt 才能理解系统。
```

例如“订单”和“客户”的关系，可能只是某个接口里的 join：

```sql
select *
from orders
join customers on orders.customer_id = customers.customer_id;
```

“创建处理任务”这个动作，可能只是一个接口：

```text
POST /tasks
```

前端知道什么时候显示按钮，后端知道怎么写任务表，但系统整体并没有显式声明：

```text
Order 是什么；
Order 和 Customer 有什么关系；
Order 能执行哪些动作；
谁能执行这些动作；
AI 应该如何围绕 Order 组织上下文。
```

Ontology 系统的关键不同在这里。

它不是只写一个订单风险应用，而是把业务世界抽成一层可复用的模型：

```text
对象类型：Order、Customer、Product、Part、Supplier、Inventory、Risk、Task
对象属性：订单号、交付日期、数量、风险等级、负责人
对象关系：Order -> Customer，Order -> Product，Product -> Part
对象动作：CreateMitigationTask、ChangePriority、RequestInventoryTransfer
权限规则：谁能看对象，谁能看字段，谁能执行动作
AI 上下文：围绕对象、关系、状态和动作自动生成
```

也就是说，普通业务系统是“页面和接口中心”；Ontology 系统是“业务对象中心”。

页面、报表、自动化流程、AI 助手，都消费同一套对象模型。

## 二、本体层的优势是什么

Ontology 的优势不是让你少写一个 CRUD 页面，而是把业务语义从具体应用里抽出来，变成可复用、可组合、可治理的一层。

### 1. 统一业务语言

企业里不同系统可能对同一件事有不同叫法：

```text
ERP 叫 sales_order；
CRM 叫 customer_order；
仓储系统叫 fulfillment_request；
财务系统叫 invoice_source。
```

Ontology 把它们映射成业务人员理解的对象：

```text
Order
Customer
Product
Inventory
Supplier
```

这样业务人员、开发者、数据分析师和 AI 都在同一套语言里工作。

### 2. 关系变成一等公民

普通系统里，对象关系通常藏在 SQL、接口和页面逻辑里。

Ontology 里，关系本身被显式建模：

```text
Order belongs to Customer
Order contains Product
Product requires Part
Part supplied by Supplier
Part has Inventory
```

这样系统可以通用地做：

```text
关系追踪；
影响分析；
上下游依赖分析；
AI 上下文组装；
对象级权限过滤。
```

### 3. 动作绑定到对象上

普通系统里，动作通常是一个按钮加一个接口。

Ontology 里，动作属于对象模型：

```text
Order 可以创建处理任务；
Order 可以调整优先级；
Part 可以发起库存调拨；
Supplier 可以发送加急请求。
```

每个动作都有明确的：

```text
作用对象；
参数 schema；
执行条件；
权限要求；
副作用；
审计记录。
```

这使系统不只是“能看数据”，而是“能围绕对象行动”。

### 4. 权限和审计下沉到对象层

普通系统里，权限经常散落在不同接口里。

Ontology 更理想的做法是把权限绑定到：

```text
对象；
字段；
关系；
动作。
```

例如：

```text
销售可以看自己客户的订单，但不能调整排产；
供应链经理可以创建风险处理任务；
财务可以看发票字段，但不能看生产细节；
AI 只能调用当前用户有权限执行的动作。
```

这对企业级 AI 尤其重要，因为 AI 不能绕过人的权限。

## 三、AI 的优势在哪里体现

AI 的优势不是替代普通业务系统，也不是替你写 CRUD。

如果只是查询订单、展示库存、创建任务，普通系统更稳定。

AI 的优势体现在四件事上。

### 1. 自然语言查询

用户可以问：

```text
未来两周哪些战略客户订单有延期风险？
```

AI 可以借助 Ontology 知道：

```text
战略客户来自 Customer.tier；
订单来自 Order；
订单和客户通过 Customer 关系连接；
延期风险来自 Risk；
风险原因可能涉及 Part、Inventory、Supplier。
```

没有 Ontology，AI 面对的是表名、字段名和接口名，很难可靠理解业务含义。

### 2. 沿关系解释原因

报表可能只显示：

```text
SO-1001 风险高
```

AI 可以沿对象关系解释：

```text
SO-1001 属于战略客户 A 公司，订单需要 500 个工业控制器。
每个工业控制器需要 1 个芯片 A，所以需要 500 个芯片 A。
当前芯片 A 可用库存只有 200 个，缺口 300 个。
供应商甲交期 14 天，因此存在延期风险。
```

这个解释不是自由发挥，而是来自对象链条。

### 3. 在合法动作空间里给建议

如果 Ontology 里定义了动作：

```text
CreateMitigationTask
RequestInventoryTransfer
NotifyAccountOwner
SendSupplierExpediteRequest
```

AI 就不是随便建议“你可以想办法处理”，而是在已有动作空间里组合：

```text
先创建处理任务；
如果其他仓库有库存，发起库存调拨；
如果没有替代库存，联系供应商加急；
如果仍然可能延期，通知客户负责人。
```

### 4. 协助执行但不绕过权限

AI 可以说：

```text
我可以为订单 SO-1001 创建一个处理任务，负责人建议为供应链经理。是否执行？
```

用户确认后，AI 调用 Ontology Action。

关键是：

```text
AI 只能执行模型里定义过的动作；
AI 只能执行当前用户有权限的动作；
动作参数有 schema 校验；
执行结果有审计记录。
```

这才是 AI + Ontology 的真正价值。

## 四、MVP 场景：订单延期风险分析

本文的 MVP 场景固定为：

> 用户打开一个订单，看到它的客户、产品、零件、库存和供应商；系统判断延期风险；AI 解释原因；用户创建一个处理任务。

最终用户路径是：

```text
进入订单列表；
找到高风险订单；
打开订单详情；
看到客户、产品、零件、库存、供应商；
理解为什么有延期风险；
让 AI 解释风险；
创建一个处理任务；
在任务列表里看到任务。
```

最小对象：

```text
Customer    客户
Order       订单
Product     产品
Part        零件
Supplier    供应商
Inventory   库存
Risk        风险
Task        处理任务
```

最小关系：

```text
Customer -> Order
Order -> Product
Product -> Part
Part -> Supplier
Part -> Inventory
Order -> Risk
Order -> Task
```

下面开始讲三条实现路线。

# 路线 A：使用 Palantir 官方平台一步一步搭 MVP

这一条路线假设你已经有 Palantir Foundry / Ontology / AIP 环境。

官方做法的核心顺序是：

```text
Dataset → Object Type → Link Type → Risk Transform → Action Type → Workshop → AIP
```

下面拆成可执行步骤。

## A1. 准备 6 个 CSV 数据源

先不要接真实 ERP。MVP 用 CSV 最快。

### customers.csv

```csv
customer_id,name,tier,owner
C001,A 公司,strategic,张三
C002,B 公司,normal,李四
```

### orders.csv

```csv
order_id,customer_id,product_id,quantity,due_date,status,priority
SO-1001,C001,P001,500,2026-08-20,in_production,high
SO-1002,C002,P002,100,2026-08-25,planned,normal
```

### products.csv

```csv
product_id,name
P001,工业控制器
P002,传感器模块
```

### parts.csv

```csv
part_id,product_id,name,required_quantity_per_product,supplier_id
PART-A,P001,芯片 A,1,S001
PART-B,P001,外壳 B,1,S002
PART-C,P002,传感器 C,1,S003
```

### suppliers.csv

```csv
supplier_id,name,lead_time_days,reliability_score
S001,供应商甲,14,0.82
S002,供应商乙,5,0.95
S003,供应商丙,7,0.90
```

### inventory.csv

```csv
part_id,available_quantity,reserved_quantity
PART-A,200,50
PART-B,1000,100
PART-C,500,50
```

产出：6 个原始数据文件。

验收：至少能看出 `SO-1001` 需要 500 个产品，而 `PART-A` 库存只有 200。

## A2. 在 Foundry 创建 Dataset

在 Foundry 里依次上传 6 个 CSV，创建 Dataset：

```text
customers
orders
products
parts
suppliers
inventory
```

检查字段类型：

```text
customer_id: string
order_id: string
product_id: string
part_id: string
quantity: integer
due_date: date
lead_time_days: integer
reliability_score: double
available_quantity: integer
```

产出：6 个 Foundry Dataset。

验收：每个 Dataset 都能预览；主键字段没有空值；`orders.due_date` 是日期类型，不是普通字符串。

## A3. 创建 Object Types

进入 Ontology 管理界面，创建对象类型。

### Customer

```text
Object Type: Customer
Dataset: customers
Primary Key: customer_id
Display Name: name
Properties:
- customer_id
- name
- tier
- owner
```

### Order

```text
Object Type: Order
Dataset: orders
Primary Key: order_id
Display Name: order_id
Properties:
- order_id
- customer_id
- product_id
- quantity
- due_date
- status
- priority
```

### Product

```text
Object Type: Product
Dataset: products
Primary Key: product_id
Display Name: name
Properties:
- product_id
- name
```

### Part

```text
Object Type: Part
Dataset: parts
Primary Key: part_id
Display Name: name
Properties:
- part_id
- product_id
- name
- required_quantity_per_product
- supplier_id
```

### Supplier

```text
Object Type: Supplier
Dataset: suppliers
Primary Key: supplier_id
Display Name: name
Properties:
- supplier_id
- name
- lead_time_days
- reliability_score
```

### Inventory

```text
Object Type: Inventory
Dataset: inventory
Primary Key: part_id
Display Name: part_id
Properties:
- part_id
- available_quantity
- reserved_quantity
```

产出：6 个 Object Types。

验收：在 Ontology 里搜索 `SO-1001`，能打开一个 Order 对象并看到属性。

## A4. 创建 Link Types

创建对象关系。

### Customer has Orders

```text
From: Customer.customer_id
To: Order.customer_id
Cardinality: one-to-many
```

### Order contains Product

```text
From: Order.product_id
To: Product.product_id
Cardinality: many-to-one
```

### Product requires Parts

```text
From: Product.product_id
To: Part.product_id
Cardinality: one-to-many
```

### Part supplied by Supplier

```text
From: Part.supplier_id
To: Supplier.supplier_id
Cardinality: many-to-one
```

### Part has Inventory

```text
From: Part.part_id
To: Inventory.part_id
Cardinality: one-to-one
```

产出：5 个 Link Types。

验收：打开 `SO-1001`，能沿关系看到：

```text
Order SO-1001
→ Customer A 公司
→ Product 工业控制器
→ Parts 芯片 A、外壳 B
→ Supplier 供应商甲、供应商乙
→ Inventory PART-A 可用 200
```

## A5. 用 Transform 生成 Risk Dataset

创建一个 Transform / Pipeline，输出 `order_risks`。

输出字段：

```text
risk_id
order_id
risk_level
risk_reason
affected_part_id
shortage_quantity
estimated_delay_days
```

计算逻辑：

```text
required_total = order.quantity * part.required_quantity_per_product
shortage = required_total - inventory.available_quantity
remaining_days = order.due_date - today

如果 shortage > 0 且 supplier.lead_time_days > remaining_days：high
如果 shortage > 0：medium
否则：low
```

伪 SQL：

```sql
select
  concat(o.order_id, '-', p.part_id) as risk_id,
  o.order_id,
  case
    when o.quantity * p.required_quantity_per_product > i.available_quantity
     and s.lead_time_days > datediff(o.due_date, current_date)
    then 'high'
    when o.quantity * p.required_quantity_per_product > i.available_quantity
    then 'medium'
    else 'low'
  end as risk_level,
  concat(
    p.name,
    ' 库存不足，缺口 ',
    cast(o.quantity * p.required_quantity_per_product - i.available_quantity as string),
    '；供应商 ',
    s.name,
    ' 交期 ',
    cast(s.lead_time_days as string),
    ' 天。'
  ) as risk_reason,
  p.part_id as affected_part_id,
  greatest(o.quantity * p.required_quantity_per_product - i.available_quantity, 0) as shortage_quantity,
  greatest(s.lead_time_days - datediff(o.due_date, current_date), 0) as estimated_delay_days
from orders o
join parts p on o.product_id = p.product_id
join inventory i on p.part_id = i.part_id
join suppliers s on p.supplier_id = s.supplier_id;
```

产出：`order_risks` Dataset。

验收：`SO-1001` 至少有一条 high 或 medium risk，原因里包含 `芯片 A` 和 `缺口 300`。

## A6. 创建 Risk Object Type，并连接 Order

创建对象：

```text
Object Type: Risk
Dataset: order_risks
Primary Key: risk_id
Display Name: risk_level
Properties:
- risk_id
- order_id
- risk_level
- risk_reason
- affected_part_id
- shortage_quantity
- estimated_delay_days
```

创建关系：

```text
Order has Risks
From: Order.order_id
To: Risk.order_id
Cardinality: one-to-many
```

产出：Risk 对象和 Order -> Risk 关系。

验收：打开 `SO-1001`，能看到风险对象，风险原因可读。

## A7. 创建 Task Dataset / Object Type

创建用于动作写入的任务数据集或对象类型。

字段：

```text
task_id
order_id
title
assignee
status
created_at
```

创建 Object Type：

```text
Object Type: Task
Primary Key: task_id
Display Name: title
Properties:
- task_id
- order_id
- title
- assignee
- status
- created_at
```

创建关系：

```text
Order has Tasks
From: Order.order_id
To: Task.order_id
Cardinality: one-to-many
```

产出：Task Object Type。

验收：可以手工创建一条 Task 测试，并从 Order 反查到 Task。

## A8. 创建 Action Type：CreateMitigationTask

在 Ontology 中创建 Action Type。

```text
Action Type: CreateMitigationTask
Applies To: Order
Purpose: 为高风险订单创建处理任务
```

参数：

```text
order: Order
assignee: string
description: string
```

默认值：

```text
title = 处理订单 {order.order_id} 的延期风险
status = open
created_at = now()
```

提交条件：

```text
Order 必须存在；
assignee 不能为空；
当前用户必须有 supply_chain_manager 角色；
如果订单没有 high/medium risk，可以给 warning 或禁止提交。
```

副作用：

```text
创建 Task 对象；
关联到当前 Order；
记录 action audit。
```

产出：一个可在 Order 上执行的 Action。

验收：打开 `SO-1001`，执行 `CreateMitigationTask`，生成一条 Task，且 Order -> Task 关系能看到它。

## A9. 配置权限

最小角色：

```text
viewer：能看 Customer、Order、Product、Part、Supplier、Inventory、Risk；
supply_chain_manager：继承 viewer，并能执行 CreateMitigationTask；
admin：能管理 Object Type / Link Type / Action Type。
```

至少配置：

```text
Order: viewer 可读；
Risk: viewer 可读；
Task: viewer 可读；
CreateMitigationTask: supply_chain_manager 可执行。
```

产出：对象和动作权限。

验收：viewer 可以看风险，但看不到或不能执行 `CreateMitigationTask`；supply_chain_manager 可以执行。

## A10. 用 Workshop 搭 3 个页面

### 页面 1：Orders

数据源：Order 对象集。

列：

```text
order_id
customer.name
due_date
status
priority
risk_level
```

交互：点击订单进入详情页。

### 页面 2：Order Detail

展示区域：

```text
订单基本信息；
客户信息；
产品信息；
零件列表；
库存情况；
供应商情况；
风险原因；
关联任务。
```

动作按钮：

```text
CreateMitigationTask
```

### 页面 3：Tasks

数据源：Task 对象集。

列：

```text
task_id
order_id
title
assignee
status
created_at
```

产出：一个最小业务应用。

验收路径：

```text
Orders 页面打开 SO-1001；
Order Detail 看到芯片 A 库存不足；
点击 CreateMitigationTask；
Tasks 页面出现新任务。
```

## A11. 接入 AIP

在 AIP 中创建一个面向供应链运营的助手。

它只做三件事：

```text
解释订单风险；
总结影响对象；
建议下一步动作。
```

给它的上下文来自 Ontology：

```text
当前 Order；
linked Customer；
linked Product；
linked Parts；
linked Suppliers；
linked Inventory；
linked Risks；
当前用户可执行 Actions。
```

示例指令：

```text
你是供应链运营助手。只基于 Ontology 上下文回答，不要编造系统里没有的数据。
回答订单延期风险时，必须说明：风险等级、主要原因、影响对象、建议动作。
如果用户要求执行动作，只能调用当前用户有权限的 Action。
```

产出：AIP 助手。

验收：问它：

```text
为什么 SO-1001 有延期风险？
```

它应该回答：

```text
因为 SO-1001 需要 500 个芯片 A，但可用库存只有 200，缺口 300；供应商甲交期 14 天，因此存在延期风险。
```

再问：

```text
下一步怎么做？
```

它应该建议执行 `CreateMitigationTask`，而不是编造一个系统里不存在的动作。

# 路线 B：使用 Palantir，但不用官方 OSDK，直接 REST API

这一条路线适合：Ontology 已经建在 Foundry 里，但你不想用 Palantir 官方 OSDK，而是自己写前端或后端，直接调 HTTP API。

整体结构：

```text
自建前端 / 后端
  ↓ REST API
Palantir Ontology API
  ↓
Object Types / Link Types / Action Types
```

## B1. 准备 API 访问凭证

你需要准备：

```text
Foundry base URL
Access token
Ontology identifier / rid
Object Type API name
Action Type API name
```

这些值在不同组织环境里命名可能不同，但最小需要知道：

```text
我要访问哪个 Ontology；
我要读哪个 Object Type；
我要对哪个对象执行哪个 Action；
当前 token 对这些对象和动作有没有权限。
```

产出：一组环境变量。

```bash
export FOUNDRY_BASE_URL="https://your-foundry.example.com"
export FOUNDRY_TOKEN="..."
export ONTOLOGY="your-ontology-api-name-or-rid"
```

验收：用 token 调一个最简单的 list ontology / metadata 接口能返回 200。

## B2. 读取 Object Type 元数据

先不要直接写业务页面。第一步先读元数据。

目标：拿到 `Order` 的属性、主键、可用 link、可用 action。

伪请求：

```bash
curl -H "Authorization: Bearer $FOUNDRY_TOKEN" \
  "$FOUNDRY_BASE_URL/api/v2/ontologies/$ONTOLOGY/objectTypes/Order"
```

你要确认返回里至少能看到：

```text
Order 的 properties；
Order 的 primary key；
Order 的 links；
Order 可执行的 actions。
```

产出：前端或后端可以动态知道 Order 是什么。

验收：不要手写“Order 有哪些字段”；先从 API 读出来。

## B3. 查询订单列表

伪请求：

```bash
curl -H "Authorization: Bearer $FOUNDRY_TOKEN" \
  "$FOUNDRY_BASE_URL/api/v2/ontologies/$ONTOLOGY/objects/Order"
```

查询条件可以先简单：

```text
limit 50；
按 due_date 或 risk_level 排序；
只取 order_id、customer_id、due_date、status、priority。
```

产出：自建应用里的订单列表。

验收：页面能显示 `SO-1001`。

## B4. 查询单个订单对象

伪请求：

```bash
curl -H "Authorization: Bearer $FOUNDRY_TOKEN" \
  "$FOUNDRY_BASE_URL/api/v2/ontologies/$ONTOLOGY/objects/Order/SO-1001"
```

产出：订单详情基础信息。

验收：返回里有：

```text
order_id = SO-1001
quantity = 500
due_date = 2026-08-20
```

## B5. 查询 linked objects

接下来不要自己写 SQL join，而是通过 Ontology Link 查询关联对象。

要查：

```text
Order -> Customer
Order -> Product
Product -> Parts
Part -> Supplier
Part -> Inventory
Order -> Risks
Order -> Tasks
```

伪请求：

```bash
curl -H "Authorization: Bearer $FOUNDRY_TOKEN" \
  "$FOUNDRY_BASE_URL/api/v2/ontologies/$ONTOLOGY/objects/Order/SO-1001/links/order_product"
```

然后继续沿 Product 查 Parts，沿 Part 查 Supplier 和 Inventory。

产出：一个由 REST API 组装出来的 Order Graph。

验收：自建后端返回统一结构：

```json
{
  "order": {},
  "customer": {},
  "product": {},
  "parts": [],
  "risks": [],
  "tasks": []
}
```

## B6. 在自建后端做 Context Builder

不要让前端直接到处调 Foundry API。建议加一层自己的后端：

```text
GET /api/orders/{order_id}/context
```

后端做：

```text
1. 读 Order；
2. 读 linked Customer；
3. 读 linked Product；
4. 读 linked Parts；
5. 读 linked Supplier / Inventory；
6. 读 linked Risks；
7. 读当前用户可执行 Actions；
8. 返回一个给页面和 AI 共用的结构。
```

产出：统一上下文接口。

验收：页面和 AI 都用这个接口，不各自拼数据。

## B7. 执行 Action Type

当用户点击“创建处理任务”时，不要直接写自己的任务表，而是调用 Foundry Ontology Action。

伪请求：

```bash
curl -X POST \
  -H "Authorization: Bearer $FOUNDRY_TOKEN" \
  -H "Content-Type: application/json" \
  "$FOUNDRY_BASE_URL/api/v2/ontologies/$ONTOLOGY/actions/CreateMitigationTask/apply" \
  -d '{
    "parameters": {
      "order": "SO-1001",
      "assignee": "供应链经理",
      "description": "检查芯片 A 替代库存或联系供应商加急"
    }
  }'
```

产出：Task 对象在 Foundry 里被创建。

验收：重新查询 `Order -> Tasks`，能看到新任务。

## B8. 自建 AI 接口，但上下文来自 Ontology API

接口：

```text
POST /api/assistant/orders/{order_id}/explain-risk
```

后端流程：

```text
1. 调 /api/orders/{order_id}/context；
2. 将 context 传给 LLM；
3. 要求 LLM 只基于 context 回答；
4. 如果建议动作，只能从 available_actions 里选择。
```

产出：不用 OSDK 的 AI 解释层。

验收：AI 解释包含真实对象链路，并且不会建议不存在或无权限的动作。

# 路线 C：完全不用 Palantir，自建轻量 Ontology MVP

这一条路线适合学习和原型验证。

目标不是复制 Palantir，而是做一个最小 Ontology Runtime：

```text
PostgreSQL：存对象数据；
Object Registry：登记对象类型；
Link Registry：登记对象关系；
Action Registry：登记动作类型；
FastAPI：提供对象、关系、风险、动作 API；
Permission Middleware：过滤字段和动作；
Action Log：记录审计；
Context Builder：给 AI 生成结构化上下文；
React / Next.js：展示最小页面；
LLM API：解释风险、生成建议。
```

## C1. 创建项目目录

```bash
mkdir ontology-mvp
cd ontology-mvp
mkdir -p backend/app backend/registry frontend
```

目录结构：

```text
ontology-mvp/
├── backend/
│   ├── app/
│   │   ├── main.py
│   │   ├── db.py
│   │   ├── seed.py
│   │   ├── risk.py
│   │   ├── context_builder.py
│   │   ├── permissions.py
│   │   └── actions.py
│   ├── registry/
│   │   ├── object_types.json
│   │   ├── link_types.json
│   │   └── action_types.json
│   └── requirements.txt
├── frontend/
└── docker-compose.yml
```

产出：项目骨架。

验收：目录存在。

## C2. 启动 PostgreSQL

`docker-compose.yml`：

```yaml
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: ontology_mvp
      POSTGRES_USER: ontology
      POSTGRES_PASSWORD: ontology
    ports:
      - "5432:5432"
    volumes:
      - ontology_pg:/var/lib/postgresql/data

volumes:
  ontology_pg:
```

启动：

```bash
docker compose up -d
```

验收：

```bash
docker ps
```

能看到 postgres 容器运行。

## C3. 创建数据库表

连接数据库：

```bash
psql "postgresql://ontology:ontology@localhost:5432/ontology_mvp"
```

建表：

```sql
create table customers (
  customer_id text primary key,
  name text not null,
  tier text not null,
  owner text not null
);

create table products (
  product_id text primary key,
  name text not null
);

create table suppliers (
  supplier_id text primary key,
  name text not null,
  lead_time_days integer not null,
  reliability_score numeric not null
);

create table parts (
  part_id text primary key,
  product_id text not null references products(product_id),
  name text not null,
  required_quantity_per_product integer not null,
  supplier_id text not null references suppliers(supplier_id)
);

create table inventory (
  part_id text primary key references parts(part_id),
  available_quantity integer not null,
  reserved_quantity integer not null
);

create table orders (
  order_id text primary key,
  customer_id text not null references customers(customer_id),
  product_id text not null references products(product_id),
  quantity integer not null,
  due_date date not null,
  status text not null,
  priority text not null
);

create table risks (
  risk_id text primary key,
  order_id text not null references orders(order_id),
  risk_level text not null,
  risk_reason text not null,
  affected_part_id text references parts(part_id),
  shortage_quantity integer,
  estimated_delay_days integer,
  created_at timestamp not null default now()
);

create table tasks (
  task_id text primary key,
  order_id text not null references orders(order_id),
  title text not null,
  assignee text not null,
  status text not null,
  created_at timestamp not null default now()
);

create table action_log (
  action_id text primary key,
  actor text not null,
  object_type text not null,
  object_id text not null,
  action_name text not null,
  parameters jsonb not null,
  created_at timestamp not null default now()
);
```

产出：业务对象表和审计表。

验收：

```sql
\dt
```

能看到 9 张表。

## C4. 插入样例数据

```sql
insert into customers values
('C001', 'A 公司', 'strategic', '张三'),
('C002', 'B 公司', 'normal', '李四');

insert into products values
('P001', '工业控制器'),
('P002', '传感器模块');

insert into suppliers values
('S001', '供应商甲', 14, 0.82),
('S002', '供应商乙', 5, 0.95),
('S003', '供应商丙', 7, 0.90);

insert into parts values
('PART-A', 'P001', '芯片 A', 1, 'S001'),
('PART-B', 'P001', '外壳 B', 1, 'S002'),
('PART-C', 'P002', '传感器 C', 1, 'S003');

insert into inventory values
('PART-A', 200, 50),
('PART-B', 1000, 100),
('PART-C', 500, 50);

insert into orders values
('SO-1001', 'C001', 'P001', 500, '2026-08-20', 'in_production', 'high'),
('SO-1002', 'C002', 'P002', 100, '2026-08-25', 'planned', 'normal');
```

验收：

```sql
select * from orders;
select * from inventory where part_id = 'PART-A';
```

能看到 `SO-1001` 和 `PART-A` 库存 200。

## C5. 创建 Object Registry

`backend/registry/object_types.json`：

```json
{
  "Order": {
    "table": "orders",
    "primary_key": "order_id",
    "display_name": "order_id",
    "properties": ["order_id", "customer_id", "product_id", "quantity", "due_date", "status", "priority"],
    "actions": ["create_mitigation_task"]
  },
  "Customer": {
    "table": "customers",
    "primary_key": "customer_id",
    "display_name": "name",
    "properties": ["customer_id", "name", "tier", "owner"]
  },
  "Product": {
    "table": "products",
    "primary_key": "product_id",
    "display_name": "name",
    "properties": ["product_id", "name"]
  },
  "Part": {
    "table": "parts",
    "primary_key": "part_id",
    "display_name": "name",
    "properties": ["part_id", "product_id", "name", "required_quantity_per_product", "supplier_id"]
  },
  "Supplier": {
    "table": "suppliers",
    "primary_key": "supplier_id",
    "display_name": "name",
    "properties": ["supplier_id", "name", "lead_time_days", "reliability_score"]
  },
  "Inventory": {
    "table": "inventory",
    "primary_key": "part_id",
    "display_name": "part_id",
    "properties": ["part_id", "available_quantity", "reserved_quantity"]
  },
  "Risk": {
    "table": "risks",
    "primary_key": "risk_id",
    "display_name": "risk_level",
    "properties": ["risk_id", "order_id", "risk_level", "risk_reason", "affected_part_id", "shortage_quantity", "estimated_delay_days"]
  },
  "Task": {
    "table": "tasks",
    "primary_key": "task_id",
    "display_name": "title",
    "properties": ["task_id", "order_id", "title", "assignee", "status", "created_at"]
  }
}
```

产出：对象元数据。

验收：后端能读取这个 JSON，并知道 `Order` 对应 `orders` 表。

## C6. 创建 Link Registry

`backend/registry/link_types.json`：

```json
{
  "links": [
    {
      "name": "order_customer",
      "from_type": "Order",
      "from_property": "customer_id",
      "to_type": "Customer",
      "to_property": "customer_id",
      "cardinality": "many-to-one"
    },
    {
      "name": "order_product",
      "from_type": "Order",
      "from_property": "product_id",
      "to_type": "Product",
      "to_property": "product_id",
      "cardinality": "many-to-one"
    },
    {
      "name": "product_parts",
      "from_type": "Product",
      "from_property": "product_id",
      "to_type": "Part",
      "to_property": "product_id",
      "cardinality": "one-to-many"
    },
    {
      "name": "part_supplier",
      "from_type": "Part",
      "from_property": "supplier_id",
      "to_type": "Supplier",
      "to_property": "supplier_id",
      "cardinality": "many-to-one"
    },
    {
      "name": "part_inventory",
      "from_type": "Part",
      "from_property": "part_id",
      "to_type": "Inventory",
      "to_property": "part_id",
      "cardinality": "one-to-one"
    },
    {
      "name": "order_risks",
      "from_type": "Order",
      "from_property": "order_id",
      "to_type": "Risk",
      "to_property": "order_id",
      "cardinality": "one-to-many"
    },
    {
      "name": "order_tasks",
      "from_type": "Order",
      "from_property": "order_id",
      "to_type": "Task",
      "to_property": "order_id",
      "cardinality": "one-to-many"
    }
  ]
}
```

产出：关系元数据。

验收：后端可以根据 Registry 知道 `Order -> Product -> Part -> Supplier` 怎么走。

## C7. 创建 Action Registry

`backend/registry/action_types.json`：

```json
{
  "create_mitigation_task": {
    "display_name": "创建延期风险处理任务",
    "object_type": "Order",
    "parameters": {
      "assignee": { "type": "string", "required": true },
      "description": { "type": "string", "required": true }
    },
    "required_role": "supply_chain_manager",
    "side_effect": "insert_task",
    "audit": true
  }
}
```

产出：动作元数据。

验收：AI 和页面都可以读取“Order 当前能执行 create_mitigation_task”。

## C8. 创建 FastAPI 后端

`backend/requirements.txt`：

```text
fastapi
uvicorn
psycopg[binary]
pydantic
python-dotenv
```

安装：

```bash
cd backend
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```

`backend/app/db.py`：

```python
import os
import psycopg
from psycopg.rows import dict_row

DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "postgresql://ontology:ontology@localhost:5432/ontology_mvp",
)

def query(sql, params=None):
    with psycopg.connect(DATABASE_URL, row_factory=dict_row) as conn:
        with conn.cursor() as cur:
            cur.execute(sql, params or {})
            return cur.fetchall()

def execute(sql, params=None):
    with psycopg.connect(DATABASE_URL, row_factory=dict_row) as conn:
        with conn.cursor() as cur:
            cur.execute(sql, params or {})
            try:
                return cur.fetchall()
            except psycopg.ProgrammingError:
                return []
```

产出：后端可以连接数据库。

验收：启动后端时不报数据库连接错误。

## C9. 实现订单列表和订单详情 API

`backend/app/main.py` 先写基础接口：

```python
from fastapi import FastAPI, HTTPException, Header
from app.db import query, execute
from app.risk import calculate_risk
from app.context_builder import build_order_context
from app.actions import create_mitigation_task

app = FastAPI()

@app.get("/objects/orders")
def list_orders():
    return query("""
        select o.*, c.name as customer_name
        from orders o
        join customers c on o.customer_id = c.customer_id
        order by o.due_date asc
    """)

@app.get("/objects/orders/{order_id}")
def get_order(order_id: str):
    rows = query("select * from orders where order_id = %(order_id)s", {"order_id": order_id})
    if not rows:
        raise HTTPException(status_code=404, detail="order not found")
    return rows[0]
```

启动：

```bash
uvicorn app.main:app --reload
```

验收：

```bash
curl http://localhost:8000/objects/orders
curl http://localhost:8000/objects/orders/SO-1001
```

能返回订单数据。

## C10. 实现 Order Graph API

`backend/app/context_builder.py`：

```python
from app.db import query

def build_order_graph(order_id: str):
    order = query("select * from orders where order_id = %(order_id)s", {"order_id": order_id})
    if not order:
        return None
    order = order[0]

    customer = query(
        "select * from customers where customer_id = %(id)s",
        {"id": order["customer_id"]},
    )[0]

    product = query(
        "select * from products where product_id = %(id)s",
        {"id": order["product_id"]},
    )[0]

    parts = query(
        "select * from parts where product_id = %(id)s",
        {"id": product["product_id"]},
    )

    enriched_parts = []
    for part in parts:
        supplier = query(
            "select * from suppliers where supplier_id = %(id)s",
            {"id": part["supplier_id"]},
        )[0]
        inventory = query(
            "select * from inventory where part_id = %(id)s",
            {"id": part["part_id"]},
        )[0]
        enriched_parts.append({
            **part,
            "required_total": order["quantity"] * part["required_quantity_per_product"],
            "supplier": supplier,
            "inventory": inventory,
        })

    risks = query("select * from risks where order_id = %(id)s", {"id": order_id})
    tasks = query("select * from tasks where order_id = %(id)s", {"id": order_id})

    return {
        "order": order,
        "customer": customer,
        "product": product,
        "parts": enriched_parts,
        "risks": risks,
        "tasks": tasks,
    }

def build_order_context(order_id: str, role: str):
    graph = build_order_graph(order_id)
    if graph is None:
        return None

    actions = []
    if role == "supply_chain_manager":
        actions.append({
            "name": "create_mitigation_task",
            "description": "创建延期风险处理任务",
        })

    return {
        **graph,
        "available_actions": actions,
    }
```

在 `main.py` 加：

```python
@app.get("/objects/orders/{order_id}/graph")
def get_order_graph(order_id: str):
    graph = build_order_context(order_id, role="viewer")
    if graph is None:
        raise HTTPException(status_code=404, detail="order not found")
    return graph
```

验收：

```bash
curl http://localhost:8000/objects/orders/SO-1001/graph
```

必须看到：

```text
order
customer
product
parts
parts[].supplier
parts[].inventory
```

## C11. 实现风险计算 API

`backend/app/risk.py`：

```python
from datetime import date

LEVEL_SCORE = {"low": 1, "medium": 2, "high": 3}

def calculate_risk(graph):
    order = graph["order"]
    due_date = order["due_date"]
    if isinstance(due_date, str):
        due_date = date.fromisoformat(due_date)

    remaining_days = (due_date - date.today()).days
    candidates = []

    for part in graph["parts"]:
        required = part["required_total"]
        available = part["inventory"]["available_quantity"]
        lead_time = part["supplier"]["lead_time_days"]
        shortage = max(required - available, 0)
        estimated_delay = max(lead_time - remaining_days, 0)

        if shortage > 0 and lead_time > remaining_days:
            level = "high"
            reason = f"{part['name']} 库存不足，缺口 {shortage}；供应商 {part['supplier']['name']} 交期 {lead_time} 天。"
        elif shortage > 0:
            level = "medium"
            reason = f"{part['name']} 库存不足，缺口 {shortage}。"
        else:
            level = "low"
            reason = f"{part['name']} 当前库存充足。"

        candidates.append({
            "risk_level": level,
            "risk_reason": reason,
            "affected_part_id": part["part_id"],
            "shortage_quantity": shortage,
            "estimated_delay_days": estimated_delay,
        })

    return max(candidates, key=lambda x: LEVEL_SCORE[x["risk_level"]])
```

在 `main.py` 加：

```python
@app.get("/objects/orders/{order_id}/risk")
def get_order_risk(order_id: str):
    graph = build_order_context(order_id, role="viewer")
    if graph is None:
        raise HTTPException(status_code=404, detail="order not found")
    return calculate_risk(graph)
```

验收：

```bash
curl http://localhost:8000/objects/orders/SO-1001/risk
```

返回应包含：

```text
risk_level
risk_reason
shortage_quantity
```

`shortage_quantity` 应该是 300。

## C12. 实现 Action：创建处理任务

`backend/app/permissions.py`：

```python
def require_role(role: str, required: str):
    if role != required:
        return False
    return True
```

`backend/app/actions.py`：

```python
import uuid
import json
from app.db import execute
from app.permissions import require_role


def create_mitigation_task(order_id: str, assignee: str, description: str, actor: str, role: str):
    if not require_role(role, "supply_chain_manager"):
        return {"error": "permission denied"}

    task_id = "TASK-" + uuid.uuid4().hex[:8]
    title = f"处理订单 {order_id} 的延期风险"

    execute(
        """
        insert into tasks(task_id, order_id, title, assignee, status)
        values (%(task_id)s, %(order_id)s, %(title)s, %(assignee)s, 'open')
        """,
        {
            "task_id": task_id,
            "order_id": order_id,
            "title": title,
            "assignee": assignee,
        },
    )

    execute(
        """
        insert into action_log(action_id, actor, object_type, object_id, action_name, parameters)
        values (%(action_id)s, %(actor)s, 'Order', %(object_id)s, 'create_mitigation_task', %(parameters)s)
        """,
        {
            "action_id": "ACT-" + uuid.uuid4().hex[:8],
            "actor": actor,
            "object_id": order_id,
            "parameters": json.dumps({"assignee": assignee, "description": description}),
        },
    )

    return {
        "task_id": task_id,
        "order_id": order_id,
        "title": title,
        "assignee": assignee,
        "status": "open",
    }
```

在 `main.py` 加：

```python
from pydantic import BaseModel

class CreateTaskRequest(BaseModel):
    assignee: str
    description: str

@app.post("/objects/orders/{order_id}/actions/create-mitigation-task")
def create_task(order_id: str, body: CreateTaskRequest, x_role: str = Header(default="viewer"), x_actor: str = Header(default="anonymous")):
    result = create_mitigation_task(
        order_id=order_id,
        assignee=body.assignee,
        description=body.description,
        actor=x_actor,
        role=x_role,
    )
    if "error" in result:
        raise HTTPException(status_code=403, detail=result["error"])
    return result
```

验收 1：viewer 不能执行。

```bash
curl -X POST http://localhost:8000/objects/orders/SO-1001/actions/create-mitigation-task \
  -H 'Content-Type: application/json' \
  -H 'X-Role: viewer' \
  -d '{"assignee":"供应链经理","description":"检查芯片 A 替代库存"}'
```

应返回 403。

验收 2：supply_chain_manager 可以执行。

```bash
curl -X POST http://localhost:8000/objects/orders/SO-1001/actions/create-mitigation-task \
  -H 'Content-Type: application/json' \
  -H 'X-Role: supply_chain_manager' \
  -H 'X-Actor: zhangsan' \
  -d '{"assignee":"供应链经理","description":"检查芯片 A 替代库存"}'
```

应返回 Task。

验收 3：审计记录存在。

```sql
select * from action_log order by created_at desc limit 5;
```

## C13. 实现 AI Context Builder 接口

在 `main.py` 加：

```python
@app.get("/assistant/orders/{order_id}/context")
def get_ai_context(order_id: str, x_role: str = Header(default="viewer")):
    context = build_order_context(order_id, role=x_role)
    if context is None:
        raise HTTPException(status_code=404, detail="order not found")
    context["risk"] = calculate_risk(context)
    return context
```

验收：

```bash
curl -H 'X-Role: viewer' http://localhost:8000/assistant/orders/SO-1001/context
```

`available_actions` 应为空或不包含创建任务。

```bash
curl -H 'X-Role: supply_chain_manager' http://localhost:8000/assistant/orders/SO-1001/context
```

`available_actions` 应包含：

```text
create_mitigation_task
```

这一步是自建版 Ontology 和普通业务系统的关键分水岭：AI 拿到的是经过权限过滤的对象上下文，而不是裸数据库。

## C14. 实现 AI 解释接口

接口：

```text
POST /assistant/orders/{order_id}/explain-risk
```

最小实现可以先不接真实 LLM，先返回模板解释；接 LLM 后，再把 context 传给模型。

模板版：

```python
@app.post("/assistant/orders/{order_id}/explain-risk")
def explain_risk(order_id: str, x_role: str = Header(default="viewer")):
    context = build_order_context(order_id, role=x_role)
    if context is None:
        raise HTTPException(status_code=404, detail="order not found")
    risk = calculate_risk(context)
    order = context["order"]
    customer = context["customer"]
    product = context["product"]

    return {
        "answer": f"订单 {order['order_id']} 的风险等级为 {risk['risk_level']}。主要原因是：{risk['risk_reason']} 该订单属于客户 {customer['name']}，产品为 {product['name']}。建议从可执行动作中选择处理方式。",
        "risk": risk,
        "available_actions": context["available_actions"],
    }
```

接 LLM 时，prompt 应该这样组织：

```text
你是供应链运营助手。
只基于给定 JSON 上下文回答，不要编造不存在的信息。
请输出：风险等级、主要原因、影响对象、建议动作。
如果建议动作，只能从 available_actions 中选择。

上下文：
{context_json}
```

验收：

```bash
curl -X POST -H 'X-Role: supply_chain_manager' \
  http://localhost:8000/assistant/orders/SO-1001/explain-risk
```

返回应包含：

```text
风险等级；
芯片 A；
缺口 300；
create_mitigation_task。
```

## C15. 做最小前端

前端只需要三个页面。

### 页面 1：订单列表

调用：

```text
GET /objects/orders
```

展示：

```text
订单号
客户
交付日期
状态
优先级
```

点击订单进入详情。

### 页面 2：订单详情

调用：

```text
GET /assistant/orders/{order_id}/context
GET /objects/orders/{order_id}/risk
```

展示：

```text
订单信息；
客户信息；
产品信息；
零件需求；
库存情况；
供应商情况；
风险原因；
可执行动作。
```

按钮：

```text
AI 解释风险
创建处理任务
```

### 页面 3：任务列表

可以先加一个接口：

```text
GET /tasks
```

展示：

```text
任务 ID
关联订单
标题
负责人
状态
创建时间
```

验收路径：

```text
打开订单列表；
进入 SO-1001；
看到芯片 A 缺口 300；
点击 AI 解释风险；
点击创建处理任务；
任务列表出现新任务。
```

## C16. 最小部署

本地 MVP 可以这样跑：

```bash
# 1. 数据库
docker compose up -d

# 2. 后端
cd backend
source .venv/bin/activate
uvicorn app.main:app --reload

# 3. 前端
cd frontend
npm run dev
```

如果先不做前端，也可以只用 Swagger：

```text
http://localhost:8000/docs
```

验收完整 API 顺序：

```bash
curl http://localhost:8000/objects/orders
curl http://localhost:8000/objects/orders/SO-1001/graph
curl http://localhost:8000/objects/orders/SO-1001/risk
curl -H 'X-Role: supply_chain_manager' http://localhost:8000/assistant/orders/SO-1001/context
curl -X POST -H 'X-Role: supply_chain_manager' -H 'Content-Type: application/json' \
  http://localhost:8000/objects/orders/SO-1001/actions/create-mitigation-task \
  -d '{"assignee":"供应链经理","description":"检查芯片 A 替代库存"}'
```

# 最小验收标准

这个 MVP 是否成立，不看页面好不好看，而看下面这些问题：

```text
1. 业务对象是否显式建模？
2. 对象关系是否显式建模？
3. 用户是否能沿关系理解风险来源？
4. 风险是否由对象关系和规则计算出来？
5. 动作是否绑定在对象上，而不是只是孤立接口？
6. 权限是否能过滤对象字段和可执行动作？
7. AI 是否通过 Context Builder 获取上下文？
8. AI 是否只能建议和调用当前用户有权限的动作？
9. 所有动作是否有审计记录？
```

如果这些问题都能回答“是”，这个 MVP 才不是普通 CRUD，而是一个真正的 Ontology MVP。

# 最后总结

普通业务系统是：

> 为一个具体流程写页面、接口和规则。

Ontology MVP 是：

> 把这个流程背后的业务对象、关系、权限和动作显式建模，让页面、流程和 AI 都复用同一套模型。

Palantir 官方路线是：

```text
Dataset → Object Type → Link Type → Risk Transform → Action Type → Workshop → AIP
```

不用官方 OSDK 但仍使用 Palantir 的路线是：

```text
Ontology API → 自建 Context Builder → 自建页面 / AI → 调用 Action API
```

完全自建路线是：

```text
PostgreSQL → Object Registry → Link Registry → Action Registry → FastAPI → Context Builder → AI → 前端
```

AI 的优势不是替你开发系统，而是：

> 基于这套模型理解业务上下文，沿对象关系解释原因，在合法动作空间里给建议，并在权限和审计约束下协助执行。

所以，最小 MVP 的正确搭建顺序是：

```text
先定义对象；
再定义关系；
再定义风险规则；
再定义动作；
再加权限和审计；
再做 Context Builder；
最后接 AI 和页面。
```

这样做出来的，才不是一个普通订单风险页面，而是一个可以继续扩展到库存、供应商、生产、客户成功等场景的本体项目起点。