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

正确顺序应该是：

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

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

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

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

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

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

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

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

普通业务系统的特点是：

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

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

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

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

POST /tasks

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

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

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

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

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

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

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

二、本体层的优势是什么

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

1. 统一业务语言

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

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

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

Order
Customer
Product
Inventory
Supplier

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

2. 关系变成一等公民

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

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

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

这样系统可以通用地做：

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

3. 动作绑定到对象上

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

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

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

每个动作都有明确的：

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

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

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

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

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

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

例如：

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

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

三、AI 的优势在哪里体现

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

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

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

1. 自然语言查询

用户可以问：

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

AI 可以借助 Ontology 知道：

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

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

2. 沿关系解释原因

报表可能只显示：

SO-1001 风险高

AI 可以沿对象关系解释：

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

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

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

如果 Ontology 里定义了动作：

CreateMitigationTask
RequestInventoryTransfer
NotifyAccountOwner
SendSupplierExpediteRequest

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

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

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

AI 可以说：

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

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

关键是：

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

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

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

本文的 MVP 场景固定为：

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

最终用户路径是：

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

最小对象：

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

最小关系：

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

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

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

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

官方做法的核心顺序是：

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

下面拆成可执行步骤。

A1. 准备 6 个 CSV 数据源

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

customers.csv

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

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

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

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

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

inventory.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：

customers
orders
products
parts
suppliers
inventory

检查字段类型：

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

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

Order

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

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

Part

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

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

Inventory

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

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

Order contains Product

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

Product requires Parts

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

Part supplied by Supplier

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

Part has Inventory

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

产出：5 个 Link Types。

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

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

A5. 用 Transform 生成 Risk Dataset

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

输出字段：

risk_id
order_id
risk_level
risk_reason
affected_part_id
shortage_quantity
estimated_delay_days

计算逻辑：

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：

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

创建对象：

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

创建关系：

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

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

字段：

task_id
order_id
title
assignee
status
created_at

创建 Object Type：

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

创建关系：

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。

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

参数：

order: Order
assignee: string
description: string

默认值：

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

提交条件：

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

副作用：

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

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

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

A9. 配置权限

最小角色：

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

至少配置：

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

产出：对象和动作权限。

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

A10. 用 Workshop 搭 3 个页面

页面 1：Orders

数据源：Order 对象集。

列：

order_id
customer.name
due_date
status
priority
risk_level

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

页面 2：Order Detail

展示区域：

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

动作按钮：

CreateMitigationTask

页面 3：Tasks

数据源：Task 对象集。

列：

task_id
order_id
title
assignee
status
created_at

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

验收路径：

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

A11. 接入 AIP

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

它只做三件事：

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

给它的上下文来自 Ontology：

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

示例指令：

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

产出：AIP 助手。

验收：问它：

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

它应该回答：

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

再问：

下一步怎么做？

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

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

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

整体结构：

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

B1. 准备 API 访问凭证

你需要准备：

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

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

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

产出：一组环境变量。

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。

伪请求：

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

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

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

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

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

B3. 查询订单列表

伪请求：

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

查询条件可以先简单：

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

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

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

B4. 查询单个订单对象

伪请求：

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

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

验收：返回里有：

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

B5. 查询 linked objects

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

要查：

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

伪请求：

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。

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

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

B6. 在自建后端做 Context Builder

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

GET /api/orders/{order_id}/context

后端做：

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。

伪请求：

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

接口：

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

后端流程：

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：

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

C1. 创建项目目录

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

目录结构：

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：

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:

启动：

docker compose up -d

验收：

docker ps

能看到 postgres 容器运行。

C3. 创建数据库表

连接数据库：

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

建表：

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()
);

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

验收：

\dt

能看到 9 张表。

C4. 插入样例数据

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');

验收：

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

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

C5. 创建 Object Registry

backend/registry/object_types.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：

{
  "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：

{
  "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：

fastapi
uvicorn
psycopg[binary]
pydantic
python-dotenv

安装：

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

backend/app/db.py：

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 先写基础接口：

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]

启动：

uvicorn app.main:app --reload

验收：

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

能返回订单数据。

C10. 实现 Order Graph API

backend/app/context_builder.py：

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 加：

@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

验收：

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

必须看到：

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

C11. 实现风险计算 API

backend/app/risk.py：

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 加：

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

验收：

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

返回应包含：

risk_level
risk_reason
shortage_quantity

shortage_quantity 应该是 300。

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

backend/app/permissions.py：

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

backend/app/actions.py：

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 加：

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 不能执行。

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 可以执行。

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：审计记录存在。

select * from action_log order by created_at desc limit 5;

C13. 实现 AI Context Builder 接口

在 main.py 加：

@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

验收：

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

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

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

available_actions 应包含：

create_mitigation_task

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

C14. 实现 AI 解释接口

接口：

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

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

模板版：

@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 应该这样组织：

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

上下文：
{context_json}

验收：

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

返回应包含：

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

C15. 做最小前端

前端只需要三个页面。

页面 1：订单列表

调用：

GET /objects/orders

展示：

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

点击订单进入详情。

页面 2：订单详情

调用：

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

展示：

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

按钮：

AI 解释风险
创建处理任务

页面 3：任务列表

可以先加一个接口：

GET /tasks

展示：

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

验收路径：

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

C16. 最小部署

本地 MVP 可以这样跑：

# 1. 数据库
docker compose up -d

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

# 3. 前端
cd frontend
npm run dev

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

http://localhost:8000/docs

验收完整 API 顺序：

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 是否成立，不看页面好不好看，而看下面这些问题：

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

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

最后总结

普通业务系统是：

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

Ontology MVP 是：

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

Palantir 官方路线是：

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

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

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

完全自建路线是：

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

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

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

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

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

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