py-acset Skill

Trit: 0 (ERGODIC - Coordinator) Self-improving Python ACSets with Ollama action space integration. Evolves on each interaction via compression progress tracking.

Overview

Python-native Attributed C-Sets with:

• Streaming Ollama integration for LLM action spaces
• ThreadPoolExecutor parallelism (configurable num_parallel)
• GF(3) load balancing across request fanout
• Self-improvement loop via interaction logging

GF(3) Triad

acsets-hatchery (-1) ⊗ py-acset (0) ⊗ gay-mcp (+1) = 0 ✓

Quick Start

# Run with flox ollama environment
flox activate -r flox/ollama -s
uv run python -m py_acset

# Or directly
uv run --with httpx python src/py_acset/main.py

Schema Definition (S-expression style)

OLLAMA_SCHEMA = {
    'objects': {'Model', 'Request', 'Response', 'Runner', 'GPU'},
    'morphisms': {
        'request_model': ('Request', 'Model'),
        'response_request': ('Response', 'Request'),
        'runner_model': ('Runner', 'Model'),
        'runner_gpu': ('Runner', 'GPU'),
    },
    'attributes': {
        'Model': {'name': str, 'size': int, 'parameters': int},
        'Request': {'id': str, 'type': str, 'prompt': str, 'status': str, 'trit': int},
        'Response': {'chunk': str, 'done': bool, 'sequence': int},
        'Runner': {'num_parallel': int, 'active': int},
        'GPU': {'memory_total': int, 'memory_free': int},
    }
}

API Endpoints Mapped

/api/generate

/api/chat

/api/embed

/api/pull

/api/tags

/api/ps

Core Classes

ACSet

from py_acset import ACSet, OLLAMA_SCHEMA

acset = ACSet(OLLAMA_SCHEMA)

# Add model
model_id = acset.add_part('Model', name='llama3.2', size=2_000_000_000)

# Add request with GF(3) trit
req_id = acset.add_part('Request', 
    id='uuid-123',
    type='generate', 
    prompt='Hello',
    trit=1,  # PLUS: generator
    model_fk=model_id)

# Query
acset.parts('Request')
acset.nparts('Response')

ParallelDispatcher

from py_acset import OllamaConfig, ParallelDispatcher

config = OllamaConfig(
    base_url="http://localhost:11434",
    num_parallel=4,
    max_queue=512
)

dispatcher = ParallelDispatcher(config)

# Streaming generate
request_id, stream = dispatcher.action_generate(
    model="llama3.2",
    prompt="Explain ACSets",
    callback=lambda chunk: print(chunk.get('response', ''), end='')
)

# Collect when done
for chunk in stream:
    if chunk is None:
        break

Parallel Fanout with GF(3)

from py_acset import parallel_fanout, collect_streams

prompts = ["Query 1", "Query 2", "Query 3"]
streams, trits = parallel_fanout(dispatcher, "llama3.2", prompts, 
                                  trit_assignment="balanced")

# trits = [-1, 0, +1] for balanced
# trits = [+1, +1, +1] for "generator"

results = collect_streams(streams, timeout=60.0)

Self-Improvement Loop

Each interaction logs to DuckDB for compression progress:

from py_acset import log_interaction, get_improvement_delta

# Automatic logging
log_interaction(
    action="generate",
    model="llama3.2", 
    tokens_in=50,
    tokens_out=200,
    latency_ms=1500
)

# Check improvement
delta = get_improvement_delta()
# Returns: {'compression_ratio': 0.85, 'latency_improvement': 0.12, ...}

Commands

# List models
just py-acset-models

# Generate (streaming)
just py-acset-generate "llama3.2" "What is category theory?"

# Parallel fanout
just py-acset-fanout "llama3.2" 3 "Explain X" "Compare Y" "Define Z"

# Self-improvement stats
just py-acset-stats

File Structure

src/py_acset/
├── __init__.py          # Exports
├── acset.py             # ACSet core
├── schema.py            # OLLAMA_SCHEMA
├── dispatcher.py        # ParallelDispatcher
├── streaming.py         # StreamChannel
├── fanout.py            # GF(3) parallel fanout
├── improvement.py       # Self-improvement tracking
└── main.py              # CLI entry point

Environment Variables

OLLAMA_HOST

localhost:11434

OLLAMA_NUM_PARALLEL

4

OLLAMA_MAX_QUEUE

512

PY_ACSET_DB

~/.topos/py_acset.duckdb

Dependencies

[project]
dependencies = [
    "httpx>=0.27",
    "duckdb>=1.0",
]

Upstream: AlgebraicJulia/py-acsets

This skill wraps and extends the official py-acsets library:

pip install git+https://github.com/AlgebraicJulia/py-acsets.git

See DISCREPANCIES.md for Julia↔Python API comparison.

Key Gaps vs ACSets.jl

rem_part!

cascading_rem_part!

copy_parts!

See Also

• acsets-hatchery

  - Julia ACSets reference

• acsets-algebraic-databases

  - Full ACSet theory

• ollama-mcp

  - MCP server for Ollama

• gay-mcp

  - Deterministic color generation

• discohy-streams

  - Categorical diagram streaming

Traversal / Mixing Random Walks

from py_acset import ACSetWalker, TriadicWalker, color_from_state

# Single walker
walker = ACSetWalker(acset, seed=1069)
for state in walker.walk(100, mode="lazy"):
    print(f"{state.ob}[{state.part_id}] trit={state.trit}")

# GF(3) triadic walkers (parallel)
triadic = TriadicWalker(acset, seed=1069)
results = triadic.mixing_race(target=50)
print(f"GF(3) conserved: {results['gf3_conserved']}")

# Deterministic color from state
color = color_from_state(state)  # e.g., "#e5a233"

Walk Modes

forward

backward

lazy

mixed

Spectral Gap

Default estimate: 1/4 (Ramanujan bound for regular graphs)

Galois Connections & Cobordant Phase Space

Subobject Classifier Ω

The 3-element Heyting algebra

{⊥, ?, ⊤}

from py_acset import SubobjectClassifier

# Characteristic morphism χ: X → Ω
chi = SubobjectClassifier.chi(
    predicate=lambda x: x.trit == 1,      # In subobject?
    boundary=lambda x: x.trit == 0         # On boundary?
)

# χ(x) returns: -1 (⊥), 0 (?), +1 (⊤)

Galois Connection: Color ⟷ Trit

from py_acset import GaloisConnection, color_to_trit_galois

galois = color_to_trit_galois(seed=1069)

# Floor (left adjoint): Color → Trit (via hue thirds)
trit = galois.floor("#e5a233")  # → -1

# Ceiling (right adjoint): Trit → Color
color = galois.ceiling(-1)  # → "#e5e533"

# Adjunction property: ⌊q⌋ ≤ p ⟺ q ≤ ⌈p⌉
assert galois.verify_adjunction(trit, "#e5a233")

Cobordant Phase Space (Pantleg)

Phase space with boundary = cobordism M: Σ_in → Σ_out

from py_acset import CobordantPhaseSpace, Port, PortInterface

# Create cobordism with colored ports
cobord = CobordantPhaseSpace("Hamiltonian", dimension=2)

# Add pantleg boundaries (in = left leg, out = right leg)
cobord.add_pantleg_port("in", Port.from_seed("q", 1069, "in"))
cobord.add_pantleg_port("in", Port.from_seed("p", 1070, "in"))
cobord.add_pantleg_port("out", Port.from_seed("q'", 1069, "out"))

# Composition requires color matching on glued boundary
other = CobordantPhaseSpace("Evolution", dimension=2)
if cobord.can_compose_with(other):
    composed = cobord.compose(other)  # M ; N

Colored Operads

Operations with colored inputs/outputs, GF(3) constraints:

from py_acset import ColoredOperad, ColoredOperation

operad = ColoredOperad("OllamaOps", seed=1069)

operad.add_operation(ColoredOperation(
    name="generate",
    input_colors=["#e5a233"],   # prompt (trit=-1)
    output_color="#33e5a2",     # response (trit=0)
))

operad.add_operation(ColoredOperation(
    name="embed",
    input_colors=["#33e5a2", "#a233e5"],  # texts
    output_color="#e5a233",                # vector
))

# Compose if colors match
result = operad.compose("embed", 0, "generate")  # plug generate into slot 0

# Subobject classifier for balanced operations
chi = operad.subobject_classifier()
print(chi("generate"))  # χ = ? (boundary) or ⊤ (balanced)

Port Graph Interface Pullback

from py_acset import PortInterface, Port

# Two interfaces can compose if ports match
interface_A = PortInterface("source")
interface_A.add_port(Port.from_seed("data", 1069, "out"))

interface_B = PortInterface("sink")
interface_B.add_port(Port.from_seed("data", 1069, "in"))  # Same seed = same color

# Pullback composition
if interface_A.can_compose(interface_B):
    # Colors match, directions compatible (out↔in)
    print("Interfaces compose!")

Categorical Structure Diagram

                    ColoredOperad
                         │
                    (operations)
                         ▼
    ┌─────────────────────────────────────┐
    │       Galois Connection             │
    │   ⌊⌋: Color → Trit (floor)          │
    │   ⌈⌉: Trit → Color (ceiling)        │
    └─────────────┬───────────────────────┘
                  │
         (lifts to subobjects)
                  ▼
    ┌─────────────────────────────────────┐
    │     SubobjectClassifier Ω           │
    │   χ: ACSet → {⊥, ?, ⊤}              │
    │   (characteristic morphism)          │
    └─────────────┬───────────────────────┘
                  │
         (classifies boundaries)
                  ▼
    ┌─────────────────────────────────────┐
    │   CobordantPhaseSpace               │
    │   ∂M = Σ_in ⊔ Σ_out                 │
    │   (pantleg = 2 boundaries)          │
    └─────────────┬───────────────────────┘
                  │
         (ports carry colors)
                  ▼
    ┌─────────────────────────────────────┐
    │      PortInterface                   │
    │   color matching via pullback        │
    │   GF(3) trit conservation           │
    └─────────────────────────────────────┘

Handoff Fidelity Boundary

Models thread transitions as cobordism compositions:

from py_acset import instantiate_current_handoff, Fidelity

# Instantiate boundary for current thread transition
boundary = instantiate_current_handoff()
report = boundary.fidelity_report()

# Classification: ⊤ (preserved), ? (at-risk), ⊥ (lost)
print(report['fidelity'])
# {'interior (⊤)': ['code', 'skills', 'duckdb'],
#  'boundary (?)': ['insights', 'state'],
#  'exterior (⊥)': ['rng', 'ephemeral']}

# Check composition viability
print(f"Can compose: {report['can_compose']}")
print(f"Matched ports: {report['matched_ports']}")

# Convert to explicit cobordism
cobord = boundary.to_cobordism()

Fidelity Categories

code

skill

duckdb

insight

state

ephemeral

Gluing via Color Matching

Σ_prev_out ≅ Σ_next_in  ⟺  ∀p: trit(p_prev) = trit(p_next)

Ports with matching colors (derived from same seed) compose. The Galois connection lifts color equality to categorical composition.

Improvement History

Self-improving on each interaction. GF(3) conserved.