akg-local 0.1.0

Agentic Knowledge Graph — solve Context Bloat in multi-agent LLM pipelines using an Atomic Dual-Brain Paradigm with zero-loss reasoning.

🧠 AKG-Local

Agentic Knowledge Graph — Solve Context Bloat in multi-agent LLM pipelines using an Atomic Dual-Brain Paradigm with zero-loss reasoning.

---

The Problem

In multi-agent LLM pipelines (PM → Dev → QA), every agent exchange bloats the context window with redundant information. Agent B re-reads everything Agent A already sent. Modify one function upstream and downstream agents silently break — logic fragmentation.

The Solution: Dual-Brain Paradigm

AKG-Local shatters every document into atomic nodes and uses two graph "brains":

Brain	Technology	Purpose
The Library (Brain 1)	Neo4j	Stores raw text nodes + DEPENDS_ON relationships
The Ledger (Brain 2)	NetworkX	In-memory HAS_READ tracking — who knows what

The Epistemic Router sits between agents: it diffs what the receiver already knows against what they need, and injects only the delta — keeping context windows flat and minimal.

When an upstream node changes, Blast Radius Invalidation traverses dependencies, forces agents to "forget" impacted nodes, and the router automatically re-injects them on the next turn.

---

Installation

pip install akg-local

Prerequisites

A running Neo4j instance:

docker run -d \
  -p 7687:7687 \
  -p 7474:7474 \
  -e NEO4J_AUTH=neo4j/password \
  neo4j

---

Quick Start

from akg_local import AKGLibrary

# Connect to Neo4j
akg = AKGLibrary(
    neo4j_uri="bolt://localhost:7687",
    auth=("neo4j", "password"),
)

# Create pipeline state with agents
state = akg.create_state(agents=["pm", "dev", "qa"])

# Ingest a Python file → atomic nodes (functions, classes, imports)
nodes = akg.ingest_file("main.py", state=state)
print(f"Shattered into {len(nodes)} atomic nodes")

# Route knowledge to an agent (zero-loss diffing)
result = akg.route(
    state,
    sender="pm",
    receiver="dev",
    target_nodes=[n.id for n in nodes],
    task_instructions="Implement the helper function.",
)
# result["messages"] contains ONLY the nodes "dev" hasn't seen yet

# Second call → dev already knows everything → empty injection
result2 = akg.route(
    state,
    sender="pm",
    receiver="dev",
    target_nodes=[n.id for n in nodes],
)
assert result2["messages"] == []  # Zero redundancy!

# Modify a node → blast radius invalidation
akg.invalidate("fn_main", state, new_text="def main():\n    print('v2')")
# All downstream dependents are now "forgotten" by every agent
# The router will auto-re-inject them on the next routing pass

akg.close()

---

Architecture

┌──────────────────────────────────────────────────────┐
│                    LangGraph Pipeline                │
│                                                      │
│  ┌─────┐     ┌──────────────────┐     ┌─────┐      │
│  │ PM  │────▶│ Epistemic Router │────▶│ Dev │      │
│  └─────┘     └────────┬─────────┘     └─────┘      │
│                       │                              │
│              ┌────────┴────────┐                    │
│              │   Delta Diff    │                    │
│              │ "What does Dev  │                    │
│              │  NOT know yet?" │                    │
│              └────────┬────────┘                    │
│                       │                              │
│         ┌─────────────┴──────────────┐              │
│         ▼                            ▼              │
│  ┌─────────────┐          ┌──────────────┐         │
│  │  Brain 1    │          │   Brain 2    │         │
│  │  Neo4j      │          │   NetworkX   │         │
│  │  (Reality)  │          │  (Epistemic) │         │
│  │             │          │              │         │
│  │ Nodes +     │◀─blast──│ HAS_READ     │         │
│  │ DEPENDS_ON  │ radius  │ edges        │         │
│  └─────────────┘          └──────────────┘         │
└──────────────────────────────────────────────────────┘

---

Core Concepts

Atomic Granularity

No more document passing. Every artifact is shattered into atomic nodes:

• A Python file → Function nodes, Class nodes, Import nodes
• A requirements doc → Paragraph nodes
• Custom text → Block nodes

Epistemic Router

from akg_local import ledger_router_node

# Drop this into your LangGraph StateGraph as a node
graph.add_node("router", ledger_router_node)

Blast Radius Invalidation

from akg_local import calculate_blast_radius_and_reset

# When fn_helper changes, find all downstream dependents and
# force every agent to forget them
state = calculate_blast_radius_and_reset(
    "fn_helper", state, library_graph
)

Agent Tools

# Give agents autonomous graph access
tools = akg.create_tools(state)
# Returns: [query_library, list_dependencies, list_dependents, update_node]

---

API Reference

AKGLibrary (Facade)

Method	Description
create_state(agents, vault)	Create a fresh AKGState
ingest_file(filepath, state)	Shatter a file into atomic nodes
ingest_text(text, source_name, state)	Ingest raw text
route(state, sender, receiver, target_nodes)	Zero-loss epistemic routing
invalidate(node_id, state, new_text)	Blast radius invalidation
create_tools(state)	Get LangChain tools for agents

AKGConfig

Factory	Description
AKGConfig.from_env()	Build from AKG_NEO4J_* env vars
AKGConfig.from_dict(d)	Build from a dictionary

Ingesters

Class	Handles
PythonIngester	.py files via AST
TextIngester	.txt, .md, .rst by paragraphs

---

Environment Variables

Variable	Default	Description
AKG_NEO4J_URI	bolt://localhost:7687	Neo4j Bolt URI
AKG_NEO4J_USER	neo4j	Neo4j username
AKG_NEO4J_PASSWORD	password	Neo4j password
AKG_BLAST_DEPTH	3	Max hops for blast radius
AKG_NEO4J_DATABASE	(default)	Neo4j database name

---

Development

# Clone
git clone https://github.com/mashhoodsiddiqui/akg-local.git
cd akg-local

# Install with dev deps
pip install -e ".[dev]"

# Run tests
pytest

# Lint
ruff check akg_local/

# Type check
mypy akg_local/

---

License

MIT — see LICENSE.