protokol-core 0.1.7

The Ultra-Lightweight, Enterprise-Grade LLM Orchestration Framework.

Protokol 🚀

The Ultra-Lightweight, Enterprise-Grade LLM Orchestration Framework for Python.

Protokol is a pure, stateless routing protocol and state machine for building complex AI Agent workflows. Unlike opinionated frameworks (like LangChain or LlamaIndex) that force you into their API wrappers, tool decorators, and networking logic, Protokol strictly handles the "Brain" (state management and routing) and leaves the execution entirely to you.

It's the ultimate Bring Your Own LLM (BYO-LLM) framework.

---

🌟 Why Protokol? (The "Bring Your Own LLM" Paradigm)

Most AI Agent frameworks tightly couple the execution logic with the routing logic. This leads to bloated dependencies, opaque API calls, and rigid schemas that break the moment you need custom infrastructure.

Protokol flips this on its head using Inversion of Control:

1. No Network Requests: Protokol has zero HTTP dependencies. It never calls an API.
2. Zero Tool Schemas: You simply yield string names of tools (e.g., ["issue_refund"]). You handle the actual execution.
3. Immutably Traceable: Every state transition is recorded in an immutable audit log (plan.trace), making it perfect for highly regulated industries (Finance, Healthcare).
4. Resumable (Human-in-the-Loop): Flows can safely pause, serialize to JSON, and be hydrated weeks later when a human manager approves a step.

When to Use Protokol ✅

• You are building enterprise AI systems (FastAPI, Celery, Kafka) where you need absolute control over API keys, network calls, and retries.
• You need Human-in-the-loop interactions that survive server restarts.
• You are building complex, multi-turn AI chatbots that require strict routing paths (e.g., Customer must provide an Order ID before advancing).
• You want to use multiple LLM providers (OpenAI for routing, local vLLM for extraction) without dealing with framework abstractions.

When NOT to Use Protokol ❌

• You want a quick script that magically connects to OpenAI and runs a web search in 3 lines of code. (Use OpenAI Assistants API or LangChain instead).
• You do not want to write your own HTTP/LLM execution loop.

---

🏗️ Core Architecture

Protokol consists of three main concepts:

1. AbstractStep: A node in the graph. It yields instructions (StepContext) and processes the result.
2. AbstractFlow: A collection of Steps. (Flows can be nested infinitely!).
3. RunPlan: The serializable memory object tracking the state, retry attempts, and the audit log.

---

📦 Installation

Protokol is published on PyPI as protokol-core and targets Python 3.10+.

python -m pip install protokol-core

Prefer using a virtual environment (e.g., python -m venv .venv && source .venv/bin/activate) or a modern tool like uv:

uv add protokol-core

Once installed, you can import every public building block via from protokol import ....

📚 Docs: Full documentation (guides + API reference) now lives at https://souvik666.github.io/protokol-core/. To build locally run pip install -e .[docs] && PYTHONPATH=src mkdocs serve.

🚢 Releases: Every push to main that follows Conventional Commits triggers semantic-release, which tags the repo, creates a GitHub release, and uploads the fresh build to PyPI.

---

✨ Minimal Example

Need a tiny script to understand the moving pieces? The example below wires a single step into a flow, asks the engine for instructions, and feeds back a mocked LLM response:

from protokol import AbstractFlow, AbstractStep, Engine, RunPlan, StepContext


class CollectGreetingStep(AbstractStep):
    id = "collect_greeting"

    def get_context(self, plan: RunPlan) -> StepContext:
        return StepContext(
            prompt="Ask the user to say hello and return 'GREETING: <text>'",
            tools=[]
        )

    def process(self, plan: RunPlan, user_result: str) -> dict:
        plan.state["greeting"] = user_result.replace("GREETING:", "").strip()
        return {"status": "success"}

    def next(self, plan: RunPlan, output: dict):
        plan.is_terminal = True
        return None


class GreetingFlow(AbstractFlow):
    id = "greeting_flow"
    steps = {"collect_greeting": CollectGreetingStep}


flow = GreetingFlow()
engine = Engine()
plan = RunPlan(current_step="collect_greeting")

context = engine.get_context(flow, plan)
print("Prompt for your executor:", context.prompt)

# Pretend your own execution layer called an LLM and received a response
mock_llm_response = "GREETING: Hi there!"
engine.advance(flow, plan, mock_llm_response)

print("Recorded state:", plan.state)

Run it after installing protokol-core; the script prints the instruction the engine generated, then shows the state captured once the mocked response is processed.

---

🔎 Component Reference

Component	What it is	How to use it	When to reach for it
AbstractStep	Base class for declarative step definitions. You implement get_context, process, and next.	Subclass it to describe a single unit of work, persist intermediate state via plan.state, and route to next steps via next().	Whenever you need deterministic orchestration around an LLM/tool call.
StepContext	Lightweight dataclass describing the prompt, tool list, and kwargs your executor needs.	Return it from get_context to tell your host application what to execute.	Every time the engine asks "what should happen next".
AbstractFlow	A collection of steps that itself behaves like a step (supports nesting).	Declare the steps mapping (ID → Step class) and instantiate the flow to wire the steps together.	For grouping related steps, composing flows, or registering sub-flows inside parent flows.
RunPlan	Serializable state machine memory including state, retries, call stack, and the typed trace.	Initialize it with a current_step, persist/load via .to_json() and .from_json() (or .to_dict() / .from_dict()), and inspect trace for auditability.	Anytime you need to pause/resume flows, inspect audit logs, or coordinate parallel steps safely.
FileStorage	Simple storage adapter implementing AbstractStorage using the filesystem.	Instantiate with a directory, then call .save(session_id, plan) / .load(session_id).	Local development, CLI demos, or lightweight deployments where a DB is not required.
Engine	Stateless runner that evaluates flows, emits contexts, and applies user results.	Instantiate once (optionally passing hooks + a RetryStrategy), call get_context before executing your LLM/tool, then call advance with the result.	When you need deterministic control flow, retries, nested flows, and audit logs without building your own state machine.
RetryStrategy (SimpleRetryStrategy, ExceptionRetryStrategy, NoRetryStrategy)	Strategy interface that decides whether a failed step should retry.	Pass a custom strategy to Engine(retry_strategy=...). Strategies receive the attempt count and raised exception.	To enforce custom retry budgets, treat certain exceptions as terminal, or disable retries entirely.
TraceEntry	Typed audit log row capturing step, attempt, result, next, and parallel metadata.	Read from plan.trace, or create custom entries via plan.add_trace_entry(...).	Compliance, analytics, and debugging workflows that require structured historical data.

🚀 Quickstart: End-to-End Usage

Here is how you build a strict conversational state machine.

1. Define Your Steps

Steps dictate what needs to happen, but they do not execute it.

from protokol import AbstractStep, RunPlan, StepContext
from typing import Union, List

class CollectNameStep(AbstractStep):
    id = "collect_name"

    def get_context(self, plan: RunPlan) -> StepContext:
        # 1. Yield instructions to your execution loop
        return StepContext(
            prompt="Ask the user for their name. If they provide it, reply 'COLLECTED: [Name]'",
            tools=[]
        )

    def process(self, plan: RunPlan, user_result: str) -> dict:
        # 2. Evaluate the LLM's response
        if "COLLECTED:" in user_result:
            plan.state["name"] = user_result.split("COLLECTED:")[1]
            return {"status": "success"}
            
        # Pause the flow if we need to ask the user a question!
        plan.is_waiting = True
        return {"status": "waiting"}

    def next(self, plan: RunPlan, output: dict) -> Union[str, List[str], None]:
        # 3. Strict Routing (The LLM cannot hallucinate past this)
        if output["status"] == "waiting":
            return "collect_name" # Loop back!
        return "welcome_user" # Advance!

2. Define Your Flow

from protokol import AbstractFlow

class OnboardingFlow(AbstractFlow):
    id = "onboarding_flow"
    steps = {
        "collect_name": CollectNameStep,
        # "welcome_user": WelcomeStep
    }

3. Your Execution Loop (The Developer's Domain)

You wrap Protokol in your own execution environment (CLI, FastAPI, etc.).

from protokol import Engine, RunPlan

flow = OnboardingFlow()
engine = Engine()
plan = RunPlan(current_step="collect_name")

while not plan.is_terminal and not plan.is_waiting:
    # 1. Ask Protokol what needs to be done
    context = engine.get_context(flow, plan)
    
    # 2. YOU execute the LLM (using any SDK or raw HTTP you want)
    llm_response = my_custom_llm_call(context.prompt) 

    # 3. Feed the result back into Protokol to advance the state machine
    engine.advance(flow, plan, llm_response)

---

💾 Persistence & Auditability

Because RunPlan is a pure dataclass with structured serialization helpers, pausing a workflow to wait for human input is incredibly simple:

from protokol import FileStorage

storage = FileStorage(directory=".sessions")

# Save state when the engine pauses
if plan.is_waiting:
    storage.save("session_123", plan)

# Hydrate state perfectly weeks later
resumed_plan = storage.load("session_123")
resumed_plan.is_waiting = False

🛠️ Advanced Features Built-In

• Typed Audit Log: Every call to advance appends a TraceEntry capturing the attempt number, result payload, next hop, and whether the step ran in parallel. Persist the plan via RunPlan.to_json() for a compliance-ready audit trail.
• Customizable Retries: Supply any RetryStrategy to the engine (e.g., SimpleRetryStrategy, ExceptionRetryStrategy, NoRetryStrategy) to decide when to re-issue the same context versus surfacing an error.
• Parallel Routing: Return a List[str] from a Step's next() method to fan-out execution. The engine preserves ordering when coordinating multiple branches.
• Nested Flows: Flows inherit from Steps, meaning a Flow can be registered as a Step inside another Flow with zero additional config.

---

🤝 Contributing

Protokol is designed to be the backbone of enterprise AI routing. Pull requests focusing on execution-agnostic state management, storage adapters (Redis, Postgres), and observability hooks are welcome!

License: MIT