How to Implement Decision Causality Analysis with Semantica’s CausalChainAnalyzer

Use the CausalChainAnalyzer class from semantica/context/causal_analyzer.py to traverse CAUSED, INFLUENCED, and PRECEDENT_FOR relationships in Semantica’s knowledge graph, supporting both direct Python API calls and MCP JSON handlers.

Semantica treats decisions as nodes in a knowledge graph and provides the CausalChainAnalyzer to trace how choices propagate through complex systems. This guide demonstrates how to implement decision causality analysis using the semantica-agi/semantica codebase, covering the analyzer’s architecture, graph backend integration, and practical querying patterns.

Understanding the CausalChainAnalyzer Architecture

The analyzer operates as a bridge between high-level causal queries and underlying graph stores. It requires only a minimal execute_query API from the backend, though it gracefully degrades to native methods when available.

Core Components

Three files define the complete causal analysis stack:

• semantica/context/causal_analyzer.py (line 72) — The CausalChainAnalyzer class providing methods for chain tracing, influence discovery, and impact scoring.
• semantica/context/decision_models.py (line 86) — The Decision dataclass that serializes node attributes and holds traversal metadata.
• mcp/tools/decisions.py (line 92) — The handle_get_causal_chain MCP handler that validates inputs and returns JSON-ready responses.

Relationship Types

The analyzer recognizes three causal edge types when traversing the graph:

• CAUSED — Direct causation between decisions
• INFLUENCED — Partial or suggestive relationships
• PRECEDENT_FOR — Historical dependency chains

Setting Up the CausalChainAnalyzer

Before tracing causal chains, obtain the active graph store from your MCP session and wire it to the analyzer.

from semantica.context import CausalChainAnalyzer
from mcp.session import get_graph

# Acquire the active graph backend (Neo4j, Amazon Neptune, etc.)

graph = get_graph()

# Instantiate the analyzer

analyzer = CausalChainAnalyzer(graph_store=graph)

The graph_store parameter accepts any backend implementing execute_query(query_string, parameters), or a graph object with a native get_causal_chain method.

Tracing Causal Chains

The primary entry point for decision causality analysis is get_causal_chain, which constructs Cypher-style queries and normalizes results into Decision objects.

Basic Upstream and Downstream Traversal

Specify the decision ID, traversal direction ("upstream" or "downstream"), and maximum depth to retrieve causal chains.

# Retrieve downstream effects of a decision

downstream = analyzer.get_causal_chain(
    decision_id="dec_12345",
    direction="downstream",
    max_depth=5
)

# Inspect results with causal distance metadata

for step in downstream:
    print(f"{step.decision_id} → distance {step.metadata['causal_distance']}")

Under the hood, get_causal_chain (lines 91-105 of causal_analyzer.py) builds a parameterized query matching the start node and recursively following causal edges up to max_depth. The private _extract_records method (lines 99-105) coerces backend-specific result shapes into flat dictionaries, while _dict_to_decision (lines 145-146) converts each record into a Decision instance enriched with causal_distance metadata.

Temporal-Aware Causal Tracing

To analyze causal chains as they existed at specific historical moments, use trace_at_time:

chain_at_time = analyzer.trace_at_time(
    event_id="dec_12345",
    at_time="2024-01-15T12:00:00Z",
    direction="downstream",
    max_depth=4
)

This method (implemented in causal_analyzer.py lines 56-68) filters the graph state to include only edges and nodes valid at the provided ISO-8601 timestamp.

Advanced Causal Analysis

Beyond simple chain traversal, the analyzer provides specialized methods for influence mapping and impact quantification.

Finding Influenced Decisions

Locate all decisions downstream-influenced by a specific choice without manually traversing the chain:

influenced = analyzer.get_influenced_decisions(
    decision_id="dec_12345",
    max_depth=10
)

print(f"{len(influenced)} decisions were downstream-influenced.")

This convenience method (lines 200-214) wraps get_causal_chain with "downstream" direction and deduplication logic.

Calculating Causal Impact Scores

Quantify a decision’s systemic importance using the impact scoring algorithm:

impact_score = analyzer.get_causal_impact_score(decision_id="dec_12345")
print(f"Impact score (0-1): {impact_score:.2f}")

The get_causal_impact_score method (lines 325-337) calculates a normalized score based on the breadth and depth of the decision’s causal footprint.

Integration with MCP (Model Context Protocol)

For JSON-API consumers, the handle_get_causal_chain function in mcp/tools/decisions.py provides a validated entry point.

from mcp.tools.decisions import handle_get_causal_chain

payload = {
    "decision_id": "dec_12345",
    "direction": "upstream",
    "max_depth": 3
}
response = handle_get_causal_chain(payload)

print("Chain length:", response["count"])

This handler (lines 92-86) validates the payload, constructs the CausalChainAnalyzer instance using the current MCP session’s graph, and returns a serializable response. If the graph backend lacks execute_query, the handler falls back to native get_causal_chain methods using signature detection (lines 15-52).

Summary

• Instantiate CausalChainAnalyzer with any graph store supporting execute_query or get_causal_chain.
• Query causal relationships using get_causal_chain with "upstream" or "downstream" directions and configurable max_depth.
• Enrich analysis with temporal constraints via trace_at_time or impact quantification via get_causal_impact_score.
• Integrate through the MCP layer using handle_get_causal_chain for standardized JSON responses.
• Reference the core implementation in semantica/context/causal_analyzer.py and data models in semantica/context/decision_models.py.

Frequently Asked Questions

What graph backends does CausalChainAnalyzer support?

The analyzer supports any backend implementing the minimal execute_query(query_string, parameters) API, including Neo4j, Amazon Neptune, and custom implementations. If the backend lacks query support but exposes a native get_causal_chain method, the analyzer automatically delegates to that method with signature adaptation.

How does the analyzer handle graph stores without Cypher support?

When execute_query is unavailable, handle_get_causal_chain (lines 15-52 in mcp/tools/decisions.py) detects the native method signature and adapts the call accordingly. This fallback mechanism ensures compatibility with graph objects that implement their own causal traversal logic.

What metadata is available in Decision objects returned by causal queries?

Each Decision object returned by get_causal_chain includes a metadata dictionary containing causal_distance (the number of hops from the source decision), normalized timestamps, and confidence values. The _dict_to_decision method (lines 145-146) handles this enrichment automatically.

Can I trace causal chains at specific historical timestamps?

Yes. Use the trace_at_time method with an ISO-8601 string or datetime object to filter the graph state temporally. This retrieves the causal chain as it existed at that specific moment, accounting for time-valid edges and node states.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}