推理与溯源

18@dataclass(frozen=True)19class DatalogFact:20    """Represents a ground truth fact."""21    predicate: str22    args: Tuple[str, ...]

242    def derive_all(self) -> List[str]:243        """244        Executes bottom-up semi-naive evaluation until fixpoint is reached.245        Returns a list of all derived facts as strings.246        """247        if self._derived:248            return [f"{f.predicate}({', '.join(f.args)})" for f in self._all_facts]249250        tracking_id = self.progress_tracker.start_tracking(251            module="reasoning",252            submodule="DatalogReasoner",253            message="Starting semi-naive fixpoint evaluation"254        )255256        iteration = 0257        newly_derived_count = 0258259        try:260            self._delta_new = self._all_facts.copy()261262            while self._delta_new:263                iteration += 1264265                # Shift deltas266                self._delta_old = self._delta_new267                self._delta_new = set()268269                delta_index = defaultdict(set)270                for f in self._delta_old:271                    delta_index[f.predicate].add(f)272273                for rule in self._rules:274                    new_facts = self._apply_rule(rule, delta_index)275276                    for fact in new_facts:277                        if fact not in self._all_facts:278                            self._delta_new.add(fact)279                            self._all_facts.add(fact)280                            self._fact_index[fact.predicate].add(fact)281                            newly_derived_count += 1282283                self.logger.debug(f"Datalog Iteration {iteration}: derived {len(self._delta_new)} new facts")284285            self._derived = True286        finally:287            self.progress_tracker.stop_tracking(288                tracking_id,289                status="completed",290                message=f"Fixpoint reached in {iteration} iterations. {newly_derived_count} new facts derived."291            )292293        return [f"{f.predicate}({', '.join(f.args)})" for f in self._all_facts]

25class Rule:26    """Simplified rule definition."""27    rule_id: str28    name: str29    conditions: List[Any]30    conclusion: Any31    rule_type: RuleType = RuleType.IMPLICATION32    confidence: float = 1.033    priority: int = 034    handler: Optional[Callable] = None35    metadata: Dict[str, Any] = field(default_factory=dict)3637@dataclass38class Fact:39    """Simple fact representation."""40    fact_id: str41    predicate: str42    arguments: List[Any]43    metadata: Dict[str, Any] = field(default_factory=dict)4445    def __str__(self) -> str:46        return f"{self.predicate}({', '.join(map(str, self.arguments))})"

64class AlphaNode(ReteNode):65    """Alpha node for single condition matching."""6667    def __init__(self, node_id: str, condition: Any):68        super().__init__(node_id)69        self.condition = condition70        self.matches: List[Fact] = []7172    def add_fact(self, fact: Fact) -> bool:73        """Add fact if it matches condition."""74        if self._matches(fact):75            self.matches.append(fact)76            return True77        return False7879    def _matches(self, fact: Fact) -> bool:80        """Check if fact matches condition."""81        # Simple matching - can be enhanced82        return True

85class BetaNode(ReteNode):86    """Beta node for joining conditions."""8788    def __init__(self, node_id: str, left: ReteNode, right: ReteNode):89        super().__init__(node_id)90        self.left = left91        self.right = right92        self.matches: List[Tuple[Fact, Fact]] = []9394    def join(self, left_fact: Fact, right_fact: Fact) -> bool:95        """Join facts from left and right nodes."""96        if self._can_join(left_fact, right_fact):97            self.matches.append((left_fact, right_fact))98            return True99        return False100101    def _can_join(self, left_fact: Fact, right_fact: Fact) -> bool:102        """Check if facts can be joined."""103        # Simple join logic - can be enhanced104        return True

107class TerminalNode(ReteNode):108    """Terminal node representing rule activation."""109110    def __init__(self, node_id: str, rule: Rule):111        super().__init__(node_id)112        self.rule = rule113        self.activations: List[Match] = []114115    def activate(self, match: Match) -> None:116        """Activate rule."""117        self.activations.append(match)

192    def _add_rule_to_network(self, rule: Rule) -> None:193        """Add rule to Rete network."""194        # Create alpha nodes for each condition195        alpha_nodes = []196        for condition in rule.conditions:197            node_id = f"alpha_{self.node_counter}"198            self.node_counter += 1199            alpha_node = AlphaNode(node_id, condition)200            alpha_nodes.append(alpha_node)201            self.network[node_id] = alpha_node202203        # Create beta nodes for joining204        if len(alpha_nodes) > 1:205            current = alpha_nodes[0]206            for i in range(1, len(alpha_nodes)):207                node_id = f"beta_{self.node_counter}"208                self.node_counter += 1209                beta_node = BetaNode(node_id, current, alpha_nodes[i])210                self.network[node_id] = beta_node211                current = beta_node212            final_node = current213        else:214            final_node = alpha_nodes[0] if alpha_nodes else None215216        # Create terminal node217        if final_node:218            node_id = f"terminal_{self.node_counter}"219            self.node_counter += 1220            terminal_node = TerminalNode(node_id, rule)221            final_node.children.append(terminal_node)222            self.network[node_id] = terminal_node

245    def _propagate_from_alpha(self, alpha_node: AlphaNode, fact: Fact) -> None:246        """Propagate from alpha node to children."""247        for child in alpha_node.children:248            if isinstance(child, BetaNode):249                # Join with matches from left side250                for left_fact in alpha_node.matches:251                    if child.join(left_fact, fact):252                        # Propagate to children253                        for grandchild in child.children:254                            if isinstance(grandchild, TerminalNode):255                                match = Match(256                                    rule=grandchild.rule,257                                    facts=[left_fact, fact],258                                    confidence=1.0,259                                )260                                grandchild.activate(match)261            elif isinstance(child, TerminalNode):262                # Direct activation263                match = Match(rule=child.rule, facts=[fact], confidence=1.0)264                child.activate(match)

154    def _save_entry(155        self,156        entry: ProvenanceEntry,157        _conn: Optional[Any] = None,158        _raise_on_error: bool = False,159    ) -> Optional[ProvenanceEntry]:160        """Compute checksum, store entry persistently, and log/handle storage errors (#783)."""161        # Hash-chain linkage (issue #825, Part A item 2): link this entry to162        # the previous entry in global insertion order before hashing, so163        # deleting a row later breaks the chain for whatever followed it.164        try:165            head = self.storage.get_chain_head(_conn)166        except Exception:167            head = None168        entry.sequence_id = (head[0] + 1) if head else 1169        entry.previous_checksum = head[1] if head else None170171        entry.checksum = compute_checksum(entry)172173        try:174            if _conn is not None:175                self.storage._store_with_conn(_conn, entry)176            else:177                self.storage.store(entry)178        except Exception as e:179            self.logger.error(180                "Failed to save provenance entry for entity '%s': %s",181                entry.entity_id,182                e,183                exc_info=True,184            )185            # Propagate when called from a batch's shared transaction so the186            # caller's per-item try/except can skip counting this item instead187            # of reporting an unpersisted entry as tracked (#807).188            if _raise_on_error:189                raise190            return None  # Graceful failure - don't return unpersisted entry (#783)191192        return entry

74    # Concatenate critical fields for checksum (entity_id intentionally excluded, see docstring)75    if isinstance(entry, dict):76        used_entities = entry.get("used_entities") or []77        data = (78            f"{entry.get('entity_type') or ''}"79            f"{entry.get('activity_id') or ''}"80            f"{entry.get('agent_id') or ''}"81            f"{entry.get('agent_type') or ''}"82            f"{entry.get('source_document') or ''}"83            f"{entry.get('timestamp') or ''}"84            f"{entry.get('confidence') if entry.get('confidence') is not None else 1.0}"85            f"{entry.get('parent_entity_id') or ''}"86            f"{entry.get('previous_version_id') or ''}"87            f"{entry.get('derived_from_id') or ''}"88            f"{','.join(used_entities)}"89            f"{entry.get('previous_checksum') or ''}"90            f"{bool(entry.get('invalidated'))}"91            f"{entry.get('invalidated_at_time') or ''}"92            f"{entry.get('invalidated_by') or ''}"93            f"{entry.get('invalidation_reason') or ''}"94        )

329                archived_history_id = None330                if existing:331                    history_entry = copy.deepcopy(existing)332                    base_history_id = f"{entity_id}:v:{existing.last_updated}"333                    history_id = base_history_id334                    counter = 1335                    while self.storage._retrieve_with_conn(conn, history_id):336                        history_id = f"{base_history_id}:{counter}"337                        counter += 1338339                    history_entry.entity_id = history_id

394                entry.previous_version_id = archived_history_id395                if explicit_parent_supplied:396                    entry.derived_from_id = parent_id

1093            entry = copy.deepcopy(existing)1094            entry.invalidated = True1095            entry.invalidated_at_time = utc_now_iso()1096            entry.invalidated_by = agent_id1097            entry.invalidation_reason = reason1098            entry.previous_version_id = history_id1099            if metadata:1100                entry.metadata = {**entry.metadata, **metadata}11011102            self._save_entry(entry, _conn=conn, _raise_on_error=True)

1256        PROV = Namespace("http://www.w3.org/ns/prov#")1257        EX = Namespace(base_uri or DEFAULT_BASE_URI)12581259        g = Graph()1260        g.bind("prov", PROV)1261        g.bind("ex", EX)12621263        def uri(entity_id: Any) -> URIRef:1264            return URIRef(EX[str(entity_id)])12651266        for e in self.storage.retrieve_all():1267            ent_uri = uri(e.entity_id)1268            g.add((ent_uri, RDF.type, PROV.Entity))12691270            if getattr(e, "timestamp", None):1271                g.add(1272                    (1273                        ent_uri,1274                        PROV.generatedAtTime,1275                        Literal(e.timestamp, datatype=XSD.dateTime),1276                    )1277                )12781279            ag_uri = None1280            if getattr(e, "agent_id", None) and e.agent_id != "unknown":1281                ag_uri = uri(e.agent_id)1282                g.add((ag_uri, RDF.type, PROV.Agent))1283                prov_subclass = self._AGENT_TYPE_PROV_CLASS.get(1284                    getattr(e, "agent_type", None)1285                )1286                if prov_subclass:1287                    g.add((ag_uri, RDF.type, PROV[prov_subclass]))1288                g.add((ent_uri, PROV.wasAttributedTo, ag_uri))12891290                # Qualified Association with hadRole (issue #825, Part A item1291                # 6): distinguishes "approved by" from "generated by" from1292                # "reviewed by" for the same agent/entity pair, which the1293                # plain wasAttributedTo triple above cannot express.1294                association = BNode()1295                g.add((ent_uri, PROV.qualifiedAssociation, association))1296                g.add((association, RDF.type, PROV.Association))1297                g.add((association, PROV.agent, ag_uri))1298                role = getattr(e, "role", None) or "generator"1299                g.add((association, PROV.hadRole, uri(f"role_{role}")))

1477        entries = sorted(1478            (e for e in self.storage.retrieve_all() if e.sequence_id is not None),1479            key=lambda e: e.sequence_id,1480        )14811482        broken_links: List[Dict[str, Any]] = []1483        expected_previous: Optional[str] = None1484        expected_sequence: Optional[int] = None1485        for entry in entries:1486            if not verify_checksum(entry):1487                broken_links.append({1488                    "entity_id": entry.entity_id,1489                    "sequence_id": entry.sequence_id,1490                    "reason": "checksum_mismatch",1491                })1492            else:1493                sequence_gap = (1494                    expected_sequence is not None1495                    and entry.sequence_id != expected_sequence + 11496                )1497                checksum_break = entry.previous_checksum != expected_previous1498                if sequence_gap or checksum_break:1499                    broken_links.append({1500                        "entity_id": entry.entity_id,1501                        "sequence_id": entry.sequence_id,1502                        "reason": "chain_break",1503                        "expected_previous_checksum": expected_previous,1504                        "actual_previous_checksum": entry.previous_checksum,1505                        "expected_sequence_id": (1506                            expected_sequence + 1 if expected_sequence is not None else None1507                        ),1508                    })15091510            # Advance state from this entry's own stored fields regardless of1511            # whether it was flagged above, so a single corrupted entry1512            # doesn't cascade into spurious breaks for every entry after it.1513            expected_previous = entry.checksum1514            expected_sequence = entry.sequence_id15151516        return {1517            "valid": len(broken_links) == 0,1518            "total_entries": len(entries),1519            "broken_links": broken_links,1520        }

24    def __init__(25        self,26        provenance: bool = False,27        agent_id: Optional[str] = None,28        is_automated: bool = True,29        **config,30    ):31        from .reasoning_engine import ReasoningEngine3233        self.provenance = provenance34        self._engine = ReasoningEngine(**config)35        self._prov_manager = None36        self._agent_id = agent_id or self.__class__.__name__37        self._is_automated = is_automated3839        if provenance:40            try:41                from semantica.provenance import ProvenanceManager42                self._prov_manager = ProvenanceManager()43            except ImportError:44                self.provenance = False

12**Reasoning vs. Retrieval:** Retrieval finds existing information that matches your query. Reasoning applies logical rules to derive new facts that weren't explicitly stored.

326**Provenance does not guarantee truth.** Provenance records faithfully track where information came from and how it was processed, but it cannot verify that the original sources were accurate. A perfectly documented chain from a flawed or malicious source still produces unreliable data.

105            if parsed_fact:106                for arg in parsed_fact.args:107                    if not arg:108                        raise ValueError("Facts cannot contain empty arguments")109                    if arg[0].isupper():110                        raise ValueError(f"Facts must be constants only. Found variable '{arg}' in {fact}")

39    # add context header40    current_context_text = f"-----{context_name}-----" + "\n"4142    # add header43    header = ["id", "text"]44    attribute_cols = (45        list(text_units[0].attributes.keys()) if text_units[0].attributes else []46    )47    attribute_cols = [col for col in attribute_cols if col not in header]48    header.extend(attribute_cols)4950    current_context_text += column_delimiter.join(header) + "\n"51    current_tokens = tokenizer.num_tokens(current_context_text)52    all_context_records = [header]

18Points supported by data should list their data references as follows:1920"This is an example sentence supported by multiple data references [Data: <dataset name> (record ids); <dataset name> (record ids)]."2122Do not list more than 5 record ids in a single reference. Instead, list the top 5 most relevant record ids and add "+more" to indicate that there are more.2324For example:2526"Person X is the owner of Company Y and subject to many allegations of wrongdoing [Data: Sources (15, 16), Reports (1), Entities (5, 7); Relationships (23); Claims (2, 7, 34, 46, 64, +more)]."2728where 15, 16, 1, 5, 7, 23, 2, 7, 34, 46, and 64 represent the id (not the index) of the relevant data record.