Extending Hellig¶
Hellig's core is a handful of Protocols and dataclasses. Every piece is designed to be swapped without forking the framework.
Custom agents¶
An Agent only needs a name and a runtime callable that maps
prompt -> str. Wrap any LLM client you like:
from hellig import Agent
def openai_runtime(prompt: str, **context):
from openai import OpenAI
client = OpenAI()
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
)
return resp.choices[0].message.content
writer = Agent(
name="writer",
instructions="You write concise, friendly prose.",
runtime=openai_runtime,
)
runtime.register_agent(writer)
instructions is automatically prepended to the prompt sent to your
callable.
Custom tools¶
Tools are plain Python functions:
def http_get(url: str) -> str:
import httpx
return httpx.get(url, timeout=10).text[:2000]
runtime.register_tool("http_get", http_get)
Tool kwargs from the stub compiler arrive as strings — cast as needed.
Plug in an LLM compiler¶
Compiler is a Protocol — implement compile(source, context) -> Program.
The standard pattern is to ask an LLM for a JSON list of instructions, then
decode it.
import json
from hellig import Compiler, Instruction, Program
SYSTEM_PROMPT = """You are the Hellig compiler. Output JSON: a list of
{"op": <op>, "args": {...}} dicts using ops say|ask|call|tool|set|done.
Available agents: {agents}. Available tools: {tools}.
"""
class LLMCompiler:
def __init__(self, llm, agents: list[str], tools: list[str]):
self.llm = llm
self.system = SYSTEM_PROMPT.format(
agents=", ".join(agents),
tools=", ".join(tools),
)
def compile(self, source: str, context=None) -> Program:
history = (context or {}).get("history", [])
plan_json = self.llm.complete(
system=self.system,
user=source,
history=history,
)
plan = json.loads(plan_json)
instrs = [Instruction(op=p["op"], args=p.get("args", {})) for p in plan]
if not instrs or instrs[-1].op != "done":
instrs.append(Instruction.done())
return Program(source=source, instructions=instrs)
Wire it into a session:
from hellig import Runtime, Session
runtime = Runtime().register_agent(writer)
session = Session(compiler=LLMCompiler(my_llm, ["writer"], []), runtime=runtime)
session.turn("Write a haiku about agents and ask me which line I like best.")
Custom IO¶
Implement say(text) and ask(prompt) -> str to deliver Hellig into a
non-terminal surface:
class WebSocketIO:
def __init__(self, ws):
self.ws = ws
def say(self, text: str) -> None:
self.ws.send_text(text)
def ask(self, prompt: str) -> str:
self.ws.send_text(prompt)
return self.ws.receive_text()
runtime = Runtime(io=WebSocketIO(ws))
Persisting sessions¶
Session is a plain dataclass — pickle it, serialize runtime.scope to
JSON, or store turn history in a database. Each Turn carries the source,
the compiled program, and the scope after the turn.
import json
state = {
"scope": session.runtime.scope,
"history": [t.user for t in session.history],
}
with open("session.json", "w") as f:
json.dump(state, f)
What lives where¶
| Component | Module | Replace by |
|---|---|---|
| Compiler | hellig.compiler |
any object with compile(source, ctx) |
| Runtime | hellig.runtime |
subclass to add new ops |
| IO | hellig.io |
implement say / ask |
| Agent | hellig.agent |
set runtime to any callable |
| Instruction set | hellig.instructions |
add a new op + _op_<name> on Runtime |