Context Graph Prototype

A Python implementation of the context graph concept from Foundation Capital's article "Context Graphs: AI's Trillion-Dollar Opportunity".

"A context graph is a living record of decision traces stitched across entities and time so precedent becomes searchable."

What is a Context Graph?

Traditional business systems capture rules (what should happen), but context graphs capture decision traces (what actually happened). This includes:

• Exceptions and overrides that deviate from standard policy
• Approval chains documenting who approved and why
• Cross-system context pulled at decision time (CRM, support, finance)
• Precedent that can be searched to inform future decisions

The key insight is that capture must happen at decision time, not retroactively via ETL. This creates a structured library of prior decisions that enables automation.

Requirements

• Python 3.10+
• No external dependencies (uses only Python standard library)

Installation

Quick Setup (Recommended)

cd context_graph_prototype
./setup_venv.sh
source .venv/bin/activate

This creates a virtual environment, installs the package, and provides the context-graph CLI command.

Manual Setup

cd context_graph_prototype
# Create virtual environment
python3 -m venv .venv
# Activate itsource .venv/bin/activate # Linux/macOS# or: .venv\Scripts\activate # Windows# Install the package
pip install -e .# Install with dev dependencies (optional)
pip install -e ".[dev]"

Without Virtual Environment

cd context_graph_prototype
python3 -m pip install -e .

Quick Start

Run the Demo

# If using venv (activate first)source .venv/bin/activate
python example.py
# Or without venv
python3 example.py

This demonstrates a deal desk scenario with:

• Sample customers and sales users
• Historical pricing decisions (discounts, exceptions, rejections)
• Precedent search
• Auto-approval based on precedent confidence
• Decision replay

Interactive CLI

# Using the installed command (requires venv activation)
context-graph
# Or run directly
python cli.py

CLI Commands

Command	Description
demo	Load sample data
search <query>	Search for precedent by text
tags <tag1> [tag2]	Search by tags
suggest <description>	Get suggested outcome based on precedent
history <entity_id>	View decision history for an entity
exceptions	List all exception decisions
record	Record a new decision interactively
replay <id>	Replay full context of a historical decision
entities	List all entities
stats	Show graph statistics
export [file]	Export graph to JSON
help	Show all available commands
quit	Exit the CLI

Project Structure

context_graph_prototype/
├── models.py # Data models (Entity, DecisionTrace, Context, Approval)
├── graph.py # Core graph storage and querying
├── search.py # Precedent search engine
├── workflow.py # Human-in-the-loop workflow engine
├── example.py # Demo scenario (deal desk pricing)
├── cli.py # Interactive CLI
├── __init__.py # Package exports
├── pyproject.toml # Python package configuration
├── requirements.txt # Dependencies (empty - stdlib only)
├── setup_venv.sh # Virtual environment setup script
├── .gitignore # Git ignore patterns
├── README.md # This file
└── docs/
└── how-it-works.md # Detailed explanation of context graphs

Documentation

• How It Works - Detailed explanation of context graphs, architecture diagrams, and the flywheel effect

Core Concepts

Entities

Business objects that decisions relate to:

• Customers, deals, contracts
• Users, products, policies

frommodelsimportEntity, EntityTypecustomer=Entity(
id="CUST-001",
type=EntityType.CUSTOMER,
name="Acme Corp",
attributes={"tier": "enterprise", "region": "NA"},
source_system="crm"
)

Decision Traces

The core unit—a recorded decision with full context:

frommodelsimportDecisionTrace, DecisionTypedecision=DecisionTrace(
type=DecisionType.APPROVAL,
description="15% discount for enterprise annual contract",
outcome="approved",
actor_id="user-123",
actor_name="Sarah Chen",
entity_ids=["CUST-001"],
tags=["discount", "enterprise", "15-percent"]
)
# Add cross-system context captured at decision timedecision.add_context("crm", {"tier": "enterprise", "arr": 150000})
decision.add_context("finance", {"payment_status": "current"})
# Record approval chaindecision.add_approval(
approver_id="mgr-456",
approver_name="Mike Johnson",
reason="Within policy limits",
channel="slack"
)

Context Graph

The graph that stores and indexes everything:

fromgraphimportContextGraphgraph=ContextGraph()
graph.add_entity(customer)
graph.record_decision(decision)
# Query capabilitieshistory=graph.get_entity_history("CUST-001")
similar=graph.find_similar(decision)
replay=graph.replay_context(decision.id)
timeline=graph.get_timeline(start=some_date, end=another_date)

Precedent Search

Find relevant past decisions:

fromsearchimportPrecedentSearchsearch=PrecedentSearch(graph)
# Text searchresults=search.search(query="discount enterprise")
# Tag searchresults=search.search(tags=["discount", "exception"])
# Combined filtersresults=search.search(
query="discount",
decision_type=DecisionType.EXCEPTION,
entity_id="CUST-001"
)
# Get suggestion for new decisionsuggestion=search.suggest_decision(
description="20% discount for startup",
entity_ids=["CUST-002"],
tags=["discount", "startup"]
)
# Returns: {"suggested_outcome": "approved", "confidence": 0.8, ...}

Workflow Engine

Human-in-the-loop automation with auto-approval:

fromworkflowimportWorkflowEngine, WorkflowRequestengine=WorkflowEngine(graph)
request=WorkflowRequest(
id="req-001",
request_type="discount_approval",
description="15% discount for enterprise customer",
requestor_id="user-123",
requestor_name="Sarah Chen",
entity_ids=["CUST-001"],
tags=["discount", "enterprise"]
)
result=engine.submit_request(request)
ifresult["auto_approved"]:
# Strong precedent found - decision was made automaticallyprint(f"Auto-approved with {result['suggestion']['confidence']:.0%} confidence")
else:
# Route for human approvaldecision=engine.approve(
request_id="req-001",
approver_id="mgr-456",
approver_name="Mike Johnson",
outcome="approved",
reason="Approved per standard policy"
)

Example Output

DEMO: Processing New Request
======================================================================
New Request: 15% discount for enterprise customer (Acme Corp)
Result: AUTO-APPROVED based on precedent!
Confidence: 80%
Based on: 6 similar decisions
Top Precedent Used:
- 15% discount request for enterprise annual contract
Outcome: approved
- 10% loyalty discount on 3-year renewal
Outcome: approved

Design Decisions

Decision	Rationale
In-memory storage	Simple for prototyping. Production would use Neo4j, PostgreSQL, or similar.
Simulated context gathering	ContextGatherer class mocks CRM/support/finance APIs. Easy to swap for real integrations.
Tag/text similarity	Simple overlap scoring. Production could use embeddings for semantic search.
No persistence	Graph resets on restart. Use export_to_json() to save state.
80% auto-approval threshold	Configurable via WorkflowEngine._auto_approve_threshold.

Extending the Prototype

Add a new entity type

# In models.pyclassEntityType(Enum):
# ... existing typesINVOICE="invoice"OPPORTUNITY="opportunity"

Add a new decision type

# In models.pyclassDecisionType(Enum):
# ... existing typesESCALATION="escalation"REFUND="refund"

Integrate real systems

# In workflow.py, extend ContextGathererclassContextGatherer:
def__init__(self):
self.salesforce=SalesforceClient()
self.zendesk=ZendeskClient()
self.stripe=StripeClient()
defgather(self, entity_ids: list[str]) ->list[Context]:
contexts= []
forentity_idinentity_ids:
# Real Salesforce integrationaccount=self.salesforce.get_account(entity_id)
ifaccount:
contexts.append(Context(
source_system="salesforce",
data=account
))
# Real Zendesk integrationtickets=self.zendesk.get_tickets(entity_id)
contexts.append(Context(
source_system="zendesk",
data={"open_tickets": len(tickets), "tickets": tickets}
))
# Real Stripe integrationcustomer=self.stripe.get_customer(entity_id)
ifcustomer:
contexts.append(Context(
source_system="stripe",
data={"balance": customer.balance, "status": customer.status}
))
returncontexts

Add persistence

# Save graph statewithopen("graph_backup.json", "w") asf:
f.write(graph.export_to_json())
# For production, integrate with a database:# - Neo4j for graph queries# - PostgreSQL with JSONB for flexibility# - MongoDB for document storage

Development

# Install dev dependencies
pip install -e ".[dev]"# Format code
black .# Lint
ruff check .# Type check
mypy .

References

• Context Graphs: AI's Trillion-Dollar Opportunity - Foundation Capital

License

MIT