OIDA 框架：为什么组织 AI 不能只靠 RAG——认知基础设施的崛起

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import math


class EpistemicCategory(Enum):
    FACT = "fact"
    DECISION = "decision"
    HYPOTHESIS = "hypothesis"
    QUESTION = "question"


@dataclass
class KnowledgeObject:
    """OIDA 知识对象：带有认知状态的结构化知识单元"""
    content: str
    category: EpistemicCategory
    importance: float = 0.5
    decay_rate: float = 0.0
    links: list[str] = field(default_factory=list)
    contradictions: list[str] = field(default_factory=list)

    # 类别特定的衰减因子
    DECAY_RATES = {
        EpistemicCategory.FACT: 0.01,      # 事实衰减最慢
        EpistemicCategory.DECISION: 0.03,  # 决策中等衰减
        EpistemicCategory.HYPOTHESIS: 0.08, # 假设衰减较快
        EpistemicCategory.QUESTION: -0.05, # 问题逆衰减(越来越重要!)
    }

    def __post_init__(self):
        self.decay_rate = self.DECAY_RATES[self.category]


class KnowledgeGravityEngine:
    """知识重力引擎：动态维护知识对象的重要性"""

    def __init__(self, max_degree: int = 7):
        self.objects: dict[str, KnowledgeObject] = {}
        self.max_degree = max_degree

    def add_object(self, obj_id: str, obj: KnowledgeObject) -> None:
        self.objects[obj_id] = obj

    def link(self, source_id: str, target_id: str) -> None:
        """创建知识对象之间的关联"""
        self.objects[source_id].links.append(target_id)
        self.objects[target_id].links.append(source_id)

    def add_contradiction(self, id1: str, id2: str) -> None:
        """标记两个知识对象之间的矛盾关系"""
        self.objects[id1].contradictions.append(id2)
        self.objects[id2].contradictions.append(id1)

    def update_gravity(self) -> None:
        """更新所有知识对象的重要性评分"""
        new_importance = {}
        for obj_id, obj in self.objects.items():
            # 基础衰减
            base = obj.importance * (1 - obj.decay_rate)
            # 邻居贡献(链接越多,重力越大)
            neighbor_boost = sum(
                self.objects[nid].importance * 0.1
                for nid in obj.links
                if nid in self.objects
            )
            # 矛盾惩罚
            contradiction_penalty = len(obj.contradictions) * 0.05
            new_importance[obj_id] = min(
                1.0, base + neighbor_boost - contradiction_penalty
            )
        for obj_id, imp in new_importance.items():
            self.objects[obj_id].importance = imp


# 使用示例
engine = KnowledgeGravityEngine()
engine.add_object("f1", KnowledgeObject(
    content="Q2 营收增长 15%", category=EpistemicCategory.FACT
))
engine.add_object("d1", KnowledgeObject(
    content="采用微服务架构", category=EpistemicCategory.DECISION
))
engine.add_object("q1", KnowledgeObject(
    content="如何降低云成本?", category=EpistemicCategory.QUESTION
))

engine.link("f1", "d1")
engine.link("f1", "q1")

engine.update_gravity()
for oid, obj in engine.objects.items():
    print(f"{oid} [{obj.category.value}]: importance={obj.importance:.3f}")
# 输出:
# f1 [fact]: importance=0.527
# d1 [decision]: importance=0.521
# q1 [question]: importance=0.576 (逆衰减使其上升!)

class EpistemicQualityScorer:
    """五组件 EQS 评估体系"""

    @staticmethod
    def completeness(objects: dict[str, KnowledgeObject]) -> float:
        """完整性: 各类别知识是否充足"""
        categories = set(o.category for o in objects.values())
        return min(1.0, len(categories) / 4.0)  # 期望 4 种类别

    @staticmethod
    def consistency(objects: dict[str, KnowledgeObject]) -> float:
        """一致性: 矛盾边越少,一致性越高"""
        total_contradictions = sum(
            len(o.contradictions) for o in objects.values()
        )
        if len(objects) == 0:
            return 1.0
        return max(0.0, 1.0 - total_contradictions / (len(objects) * 2))

    @staticmethod
    def timeliness(objects: dict[str, KnowledgeObject]) -> float:
        """时效性: 知识的重要性是否合理分布"""
        if not objects:
            return 0.0
        importances = [o.importance for o in objects.values()]
        # 重要性分布越均匀,时效性越好(方差小)
        mean = sum(importances) / len(importances)
        variance = sum((x - mean) ** 2 for x in importances) / len(importances)
        return max(0.0, 1.0 - variance)

    @staticmethod
    def traceability(objects: dict[str, KnowledgeObject]) -> float:
        """可追溯性: 对象之间是否有足够的链接"""
        total_links = sum(len(o.links) for o in objects.values())
        if len(objects) <= 1:
            return 1.0
        # 平均每个对象至少有 2 个链接为满分
        return min(1.0, total_links / (len(objects) * 2))

    @staticmethod
    def cyclicity(objects: dict[str, KnowledgeObject]) -> float:
        """循环性分析: 检测推理链中的循环依赖"""
        visited = set()
        cycles = 0
        for start_id in objects:
            if start_id in visited:
                continue
            path = set()
            stack = [start_id]
            while stack:
                current = stack.pop()
                if current in path:
                    cycles += 1
                    continue
                path.add(current)
                visited.add(current)
                for link in objects[current].links:
                    if link in objects:
                        stack.append(link)
        return max(0.0, 1.0 - cycles * 0.1)

    @classmethod
    def calculate_eqs(cls, objects: dict[str, KnowledgeObject]) -> float:
        """计算综合 EQS 评分(五组件加权平均)"""
        weights = {
            "completeness": 0.2,
            "consistency": 0.25,
            "timeliness": 0.15,
            "traceability": 0.2,
            "cyclicity": 0.2,
        }
        scores = {
            "completeness": cls.completeness(objects),
            "consistency": cls.consistency(objects),
            "timeliness": cls.timeliness(objects),
            "traceability": cls.traceability(objects),
            "cyclicity": cls.cyclicity(objects),
        }
        return sum(scores[k] * weights[k] for k in weights)


# 使用示例
objects = {
    "f1": KnowledgeObject("Q2 营收增长 15%", EpistemicCategory.FACT),
    "d1": KnowledgeObject("采用微服务架构", EpistemicCategory.DECISION),
    "q1": KnowledgeObject("如何降低云成本?", EpistemicCategory.QUESTION),
}
objects["f1"].links = ["d1"]
objects["d1"].links = ["f1"]
objects["q1"].links = ["f1"]

eqs = EpistemicQualityScorer.calculate_eqs(objects)
print(f"EQS Score: {eqs:.3f}")
# 输出: EQS Score: 0.640