prismlang 0.1.1

Deterministic vector language protocol for multi-agent AI orchestration

PrismLang

Deterministic Vector Language Protocol for LangGraph Multi-Agent AI

Stop paying the token tax on every agent hop. Start routing with math.

📖 Docs · 🚀 Quickstart · 📊 Benchmarks · 🏢 Insight IT Solutions

pip install prismlang

---

What Problem Does PrismLang Solve?

If you are building multi-agent AI systems with LangGraph, you are likely running into one or more of these problems right now:

---

Problem 1 — You are paying for the same tokens over and over

Every node in a LangGraph pipeline reads the entire message history as prompt tokens. Each agent re-reads everything every prior agent wrote — even context it doesn't need.

Turn 1 →  800 B   agent A pays for its own output
Turn 2 → 1,600 B  agent B pays for A + B
Turn 3 → 2,400 B  agent C pays for A + B + C   ← 3× cost for the same data
Turn N → N×800 B  cost grows linearly with graph depth

In a 10-node production graph running thousands of times a day, this is not a rounding error — it is a significant and avoidable infrastructure cost.

PrismLang fixes this by replacing growing text history with compact 64-number vectors. Each agent turn costs ~414 bytes regardless of graph depth.

---

Problem 2 — You have no audit trail for routing decisions

When your pipeline misroutes a request — sends a compliance question to the market-data agent, or a triage case to the wrong specialist — you have no structured record of why. You are debugging with logs or nothing.

Regulators in healthcare (HIPAA), finance (SOX, Basel III), and legal (privilege review) are increasingly asking: show us how your AI made this decision.

PrismLang fixes this by attaching a rule_chain to every agent output — a full trace of the encoding, category inference, and projection steps. Every routing decision is reproducible and explainable from first principles.

envelope["rule_chain"]
# ['text -> encoder(all-MiniLM-L6-v2, d=384)',
#  "category_inference -> slug='risk'",
#  'spherical_blend(alpha=0.300) -> v_prime',
#  "JL_reduction(seed=sha256('acme-finance'), k=64) -> p"]

---

Problem 3 — Multi-tenant AI has no safe isolation layer

In SaaS AI platforms, multiple clients share the same agents and graph infrastructure. One misconfigured node, one wrong state key, and Tenant A's reasoning context is visible to Tenant B's inference call.

Text payloads don't enforce isolation — they rely entirely on your application layer getting it right, every time.

PrismLang fixes this by making isolation mathematical. Each tenant gets a unique Johnson-Lindenstrauss projection matrix derived from SHA-256(tenant_id). The same input text produces geometrically incompatible vectors under different tenant keys. Cross-tenant leakage is provably impossible at the vector level.

---

Who is PrismLang for?

If you are building...	PrismLang helps you...
Multi-agent LangGraph pipelines	Cut token costs 57–62% without changing agent logic
Multi-tenant SaaS AI products	Add cryptographic tenant isolation at the protocol layer
Healthcare or finance AI systems	Produce a full audit trail on every routing decision
AI platforms with compliance requirements	Satisfy regulators with structured, reproducible decision records
Any LangGraph graph with 3+ nodes	Reduce state size linearly — the deeper the graph, the bigger the saving

---

The Solution

PrismLang replaces growing text payloads with 64-number deterministic vectors — one per agent turn. A single decorator on your existing nodes. No agent refactoring. No LLM retraining.

[Your Agent]  →  "Credit risk elevated in EM bonds."  (text, 400+ tokens)
                              ↓  @prism_node
              →  PrismEnvelope { vector[64], slug="risk", rule_chain }  (~414 bytes)

The math guarantees that the same input always produces the same vector, that different tenants produce incompatible vectors, and that every routing decision is traceable back to a taxonomy rule.

---

How It Works

PrismLang applies two equations on every agent output:

Step 1 — Spherical Blend (pulls the embedding toward its category direction)

v' = normalize( (1 − α) · v  +  α · ‖v‖ · eᵢ )

Step 2 — JL Reduction (compresses to k=64 dims, isolated per tenant)

p = normalize( P · v' )

Where P is a (64 × 384) Gaussian matrix seeded from SHA-256(tenant_id). A vector stolen from Tenant A is geometrically meaningless to any model operating under Tenant B's projection.

                    ┌─────────────────────────────────────────────┐
                    │           Your LangGraph Graph              │
                    │                                             │
  [researcher] ──→ [summarizer] ──→ [reviewer] ──→ [translator]  │
       │                │               │               │         │
  @prism_node      @prism_node     @prism_node     (boundary)     │
       │                │               │               │         │
  PrismEnvelope    PrismEnvelope   PrismEnvelope   Human text     │
  {64-d vector}    {64-d vector}   {64-d vector}                  │
  {rule_chain}     {rule_chain}    {rule_chain}                    │
                    │                                             │
                    │  prism_sequence  ─────────  append-only     │
                    └─────────────────────────────────────────────┘

---

Quick Start

from prismlang import (
    Category, TaxonomyConfig, PrismProjector,
    PrismState, prism_node, BoundaryTranslator,
    JsonFileCheckpointer,
)
from langgraph.graph import StateGraph, END

# 1. Define your domain taxonomy
taxonomy = TaxonomyConfig(categories=[
    Category("risk",       "Market Risk",   ["risk", "exposure", "volatility"]),
    Category("market",     "Market Data",   ["price", "equity", "bond"]),
    Category("compliance", "Compliance",    ["regulation", "audit", "kyc"]),
])

# 2. One projector per tenant — cryptographically isolated
projector = PrismProjector(taxonomy, tenant_id="acme-finance-prod", k=64)

# 3. Decorate your existing nodes — zero changes to agent logic
@prism_node(agent_id="analyst", projector=projector)
def analyst(state: PrismState) -> dict:
    return {"raw_output": "Credit risk exposure elevated in EM bonds."}

@prism_node(agent_id="reviewer", projector=projector)
def reviewer(state: PrismState) -> dict:
    prev = state["prism_sequence"][-1]["category_slug"]
    return {"raw_output": f"Reviewing {prev} findings for compliance sign-off."}

# 4. Build and run — exactly like any LangGraph graph
translator = BoundaryTranslator()
graph = StateGraph(PrismState)
graph.add_node("analyst",    analyst)
graph.add_node("reviewer",   reviewer)
graph.add_node("translator", translator.as_langgraph_node())
graph.set_entry_point("analyst")
graph.add_edge("analyst", "reviewer")
graph.add_edge("reviewer", "translator")
graph.add_edge("translator", END)

app = graph.compile(checkpointer=JsonFileCheckpointer())
result = app.invoke({
    "prism_sequence": [], "raw_output": "", "tenant_id": "acme-finance-prod"
})

# Inspect the audit envelope
envelope = result["prism_sequence"][0]
print(envelope["category_slug"])   # "risk"
print(len(envelope["vector"]))     # 64
print(envelope["rule_chain"])
# ['text -> encoder(all-MiniLM-L6-v2, d=384)',
#  "category_inference -> slug='risk'",
#  'spherical_blend(alpha=0.300) -> v_prime',
#  "JL_reduction(seed=sha256('acme-finance-prod'), k=64) -> p"]

---

Code Examples

Full examples for every use case are in EXAMPLES.md.

Multi-Node Pipeline

from prismlang import (
    Category, TaxonomyConfig, PrismProjector,
    PrismState, prism_node, BoundaryTranslator, JsonFileCheckpointer,
)
from langgraph.graph import StateGraph, END

taxonomy = TaxonomyConfig(categories=[
    Category("risk",       "Market Risk",  ["risk", "exposure", "volatility"]),
    Category("portfolio",  "Portfolio",    ["allocation", "rebalance", "weight"]),
    Category("compliance", "Compliance",   ["regulation", "audit", "kyc"]),
])
projector = PrismProjector(taxonomy, tenant_id="acme-finance", k=64)

@prism_node(agent_id="risk_analyst", projector=projector)
def risk_analyst(state: PrismState) -> dict:
    return {"raw_output": "Portfolio VaR at 99%: $4.2M. EM exposure elevated."}

@prism_node(agent_id="portfolio_manager", projector=projector)
def portfolio_manager(state: PrismState) -> dict:
    prev = state["prism_sequence"][-1]["category_slug"]  # reads previous category
    return {"raw_output": f"Reducing {prev} exposure. Rotating to investment-grade fixed income."}

@prism_node(agent_id="compliance_officer", projector=projector)
def compliance_officer(state: PrismState) -> dict:
    return {"raw_output": "Rebalance approved. SEC 13F disclosure required within 45 days."}

graph = StateGraph(PrismState)
graph.add_node("risk",       risk_analyst)
graph.add_node("portfolio",  portfolio_manager)
graph.add_node("compliance", compliance_officer)
graph.add_node("exit",       BoundaryTranslator().as_langgraph_node())
graph.set_entry_point("risk")
graph.add_edge("risk",      "portfolio")
graph.add_edge("portfolio", "compliance")
graph.add_edge("compliance", "exit")
graph.add_edge("exit", END)

app = graph.compile(checkpointer=JsonFileCheckpointer())
result = app.invoke(
    {"tenant_id": "acme-finance", "prism_sequence": [], "raw_output": ""},
    config={"configurable": {"thread_id": "run-001"}},
)

for env in result["prism_sequence"]:
    print(f"[{env['agent_id']:20}] → {env['category_slug']}")
# [risk_analyst        ] → risk
# [portfolio_manager   ] → portfolio
# [compliance_officer  ] → compliance

Multi-Tenant Isolation

import numpy as np
from prismlang import Category, TaxonomyConfig, PrismProjector

taxonomy = TaxonomyConfig(categories=[Category("risk", "Risk", ["risk"])])

proj_a = PrismProjector(taxonomy, tenant_id="hospital-a")
proj_b = PrismProjector(taxonomy, tenant_id="hospital-b")

text = "Patient shows elevated cardiac risk markers."
_, vec_a, _ = proj_a.project(text)
_, vec_b, _ = proj_b.project(text)

print(f"Cross-tenant cosine similarity: {np.dot(vec_a, vec_b):.4f}")  # ~0.15 — near-zero
print(f"Same-tenant determinism:        {np.dot(vec_a, proj_a.project(text)[1]):.4f}")  # 1.0000

Async Nodes

from prismlang import Category, TaxonomyConfig, PrismProjector, PrismState, async_prism_node
from langgraph.graph import StateGraph, END
import asyncio

taxonomy = TaxonomyConfig(categories=[Category("analysis", "Analysis", ["data", "insight"])])
projector = PrismProjector(taxonomy, tenant_id="async-org")

@async_prism_node(agent_id="async_agent", projector=projector)
async def async_agent(state: PrismState) -> dict:
    await asyncio.sleep(0)  # your async LLM call here
    return {"raw_output": "Async analysis complete. Strong upward trend detected."}

graph = StateGraph(PrismState)
graph.add_node("agent", async_agent)
graph.set_entry_point("agent")
graph.set_finish_point("agent")

result = asyncio.run(graph.compile().ainvoke({
    "tenant_id": "async-org", "prism_sequence": [], "raw_output": ""
}))
print(result["prism_sequence"][0]["category_slug"])  # analysis

Reading the Audit Trail

from prismlang import Category, TaxonomyConfig, PrismProjector

taxonomy = TaxonomyConfig(categories=[
    Category("compliance", "Compliance", ["kyc", "aml", "regulation", "audit"]),
])
projector = PrismProjector(taxonomy, tenant_id="regulated-bank-001")

slug, vector, rule_chain = projector.project("KYC review flagged three accounts for AML investigation.")

for step in rule_chain:
    print(step)
# text -> encoder(all-MiniLM-L6-v2, d=384)
# category_inference -> slug='compliance'
# spherical_blend(alpha=0.300) -> v_prime
# JL_reduction(seed=sha256('regulated-bank-001'), k=64) -> p

---

Benchmark Results

Measured against standard LangGraph text-state across three enterprise domains.
Full methodology in docs/BENCHMARK.md. Results stored in PostgreSQL.

Domain	Metric	Standard LangGraph	PrismLang	Change
🏥 Healthcare ICU triage pipeline	Prompt tokens (3 turns)	391	148	−62.1%
	State size (turn 3)	1,928 B	960 B	−50.2%
💹 Finance Risk / portfolio pipeline	Prompt tokens (3 turns)	407	175	−57.0%
	State size (turn 3)	1,760 B	960 B	−45.5%
📈 Trade Market Signal / execution pipeline	Prompt tokens (3 turns)	435	180	−58.6%
	State size (turn 3)	1,867 B	960 B	−48.6%

LLM inference latency: unchanged. PrismLang reduces state transport, not compute.
Encoding overhead per turn: ~31–35 ms CPU-only (no GPU required).

---

Key Properties

Property	Detail
Zero agent refactoring	Agents return {"raw_output": "..."} — nothing else changes
Deterministic	Same text + same tenant = identical vector, always
Full audit trail	Every envelope carries a rule_chain tracing the full decision path
Tenant isolation	SHA-256(tenant_id) seeds the JL matrix — cross-tenant vectors are incompatible
No GPU	ONNX Runtime CPU inference — runs on any standard server
No external API	Encoder is fully local — no network call per token
Model-agnostic	Works with GPT-4, Claude, Gemini, Llama, or any LLM
Async native	@async_prism_node for async LangGraph nodes
Two checkpointers	JsonFileCheckpointer (zero deps) + PostgresCheckpointer

---

Installation Options

# Core (local JSON checkpointing)
pip install prismlang

# PostgreSQL checkpointing
pip install "prismlang[postgres]"

# Async support (asyncpg + aiofiles)
pip install "prismlang[async-postgres,async-files]"

# Full development environment
pip install "prismlang[dev]"

---

Run the Benchmarks

git clone https://github.com/insightitsGit/prismlang
cd prismlang
pip install -e ".[dev]"

# Runs all 3 domain benchmarks and prints comparison table
python -m benchmarks.run_all

Requires a running PostgreSQL instance. Set DATABASE_URL or use the default:
postgresql://insight_admin:...@localhost/prismLangDB

---

Project Structure

prismlang/
├── prismlang/
│   ├── encoder.py        # ONNX all-MiniLM-L6-v2 → 384-d unit vector
│   ├── taxonomy.py       # TaxonomyConfig + Category direction vectors (eᵢ)
│   ├── projector.py      # PrismProjector: spherical blend + JL reduction
│   ├── middleware.py     # @prism_node + @async_prism_node decorators
│   ├── checkpointer.py   # JsonFile + Postgres + Async variants
│   ├── exceptions.py     # Typed exception hierarchy (17 classes)
│   ├── envelope.py       # PrismEnvelope TypedDict
│   ├── state.py          # PrismState (LangGraph append-only channel)
│   └── translator.py     # BoundaryTranslator (structural reconstruction)
├── benchmarks/
│   └── domains/          # Healthcare · Finance · Trade Market
├── demo/
│   └── graph.py          # Runnable 3-node LangGraph demo
├── tests/                # 34 tests · 0 failures
└── docs/
    ├── ARCHITECTURE.md
    ├── BENCHMARK.md
    └── SECURITY.md

---

Security

PrismLang's tenant isolation is a geometric property guaranteed by the Johnson-Lindenstrauss lemma — not an access-control system. For production deployments, see docs/SECURITY.md which covers:

• What the JL matrix does and does not protect
• Overlay encryption for PII in raw_output
• Dependency security notes (onnxruntime, psycopg2, asyncpg)
• NumPy PRNG stability across version upgrades

To report a vulnerability: [e-mail] — do not open a public GitHub issue.

---

Citation

@techreport{parva2026prismlang,
  title       = {PrismLang: A Deterministic Vector Language Protocol
                 for Auditable Multi-Agent AI Orchestration},
  author      = {Parva, Amin},
  year        = {2026},
  institution = {Insight IT Solutions LLC},
  url         = {https://www.insightits.com/prismlang}
}

---

License

Apache 2.0 — free for commercial and personal use.

---

Built by Insight IT Solutions LLC

Enterprise AI systems · LangGraph architecture · Vector search · Production deployment

🌐 Website · 📧 Contact · 🔒 Security