the chat agent

an end-to-end example that ties everything together.

We'll build a multi-turn LLM chat agent. Under 50 lines. Every idea from the book shows up: actors, state, FX effects, @zef_function, topics, HTTPRequest. It's the capstone.

what we're building

FX.Publish FX.Print User ────────────▶ topic ────▶ Actor ──────────▶ stdout │ │ holds conversation state │ calls LLM over HTTP │ appends response to history ▼ state = { history: [...], system_prompt: '...', }

the whole thing

from zef import *
import builtins, time

topic = ET.Topic('🍃-cafe000000000000a011')

@zef_function
def chat(msg: String, state: Any) -> Array:
    # append the user's turn
    history = state['history'] | append({'role': 'user', 'content': msg}) | collect

    # build the full messages list for the API
    messages = [{'role': 'system', 'content': state['system_prompt']}] + list(history)
    body = {'model': 'qwen2.5:32b', 'stream': False, 'messages': messages} | to_json | collect

    # fire the HTTP request as an FX
    resp = FX.HTTPRequest(
        url='http://localhost:11434/api/chat',
        method='POST',
        headers={'Content-Type': 'application/json'},
        body=body,
    ) | run

    # extract the reply
    reply = resp | F.body | parse_json | F.message | F.content | collect

    # append assistant turn
    history = history | append({'role': 'assistant', 'content': reply}) | collect

    # emit a print effect and store new state
    return [
        [FX.Print(content=f'🤖  {reply}')],
        {**state, 'history': history},
    ]

# start the actor
actor = FX.StartActor(
    input=topic,
    initial_state={
        'history':       [],
        'system_prompt': 'Reply warmly in one short sentence.',
    },
    handler=chat,
) | run

# REPL loop — feed user input into the topic
while (line := builtins.input('🧑  ').strip()):
    FX.Publish(target=topic, content=line) | run
    time.sleep(6.0)    # let the LLM reply before next prompt

FX.StopActor(actor=actor) | run

what's happening — line by line

the mental playback

You type "hi, I'm Kai"
FX.Publish drops it on the topic
The actor receives (msg, state) — msg = "hi, I'm Kai", state = the dict
Handler appends user turn to history
Builds the API payload
Fires FX.HTTPRequest → Ollama
Parses the response, extracts the assistant reply
Returns [[FX.Print(...)], new_state]
Runtime fires the Print; stores the new state
Next message, state now has both turns

what each zef feature buys you

feature	what it gives
Actor	private state per session — no locks, no globals
Topic	decouples "how messages arrive" from "how they're processed"
FX.HTTPRequest	provider calls are data — swap Ollama → Gemini by changing one string
FX.Print	side effect labeled as one; function stays pure-ish
@zef_function	handler is content-addressed — reproducible and shippable
ZefOps	parsing (`parse_json \| F.message \| F.content`) is a one-liner
dict state	history grows naturally; any Zef value would work too

swapping providers — the big payoff

Want Gemini instead of Ollama? Change the URL + payload:

resp = FX.HTTPRequest(
    url=f'https://generativelanguage.googleapis.com/v1beta/models/gemini-pro:generateContent?key={api_key}',
    method='POST',
    headers={'Content-Type': 'application/json'},
    body={'contents': [{'parts': [{'text': msg}]}]} | to_json | collect,
) | run

No rewrite. The actor, the topic, the state shape all stay exactly the same. That's the point of "effects as data" — provider integration is a config change, not a refactor.

enhancements

persistent history

db = FX.CreateDB(
    type=DictDatabase,
    persistence='local_disk',
    path='~/.chat/history.zef',
) | run

# in the handler, after updating history:
db['last_session'] = history

expose as an HTTP endpoint

@zef_function
def api_chat(req):
    user_msg = str(req.body) | parse_json | F.message | collect
    FX.Publish(target=topic, content=user_msg) | run
    # (return an ack; the actor handles printing/storing)
    return ET.JSON(content={'ok': True})

FX.StartHTTPServer(
    routes={('POST', '/chat'): api_chat},
    port=8000,
) | run

multiple conversation sessions

Use one actor per session. Each has its own state. Route by session ID:

sessions = {}   # session_id → (topic, actor)

# when a new user arrives:
def new_session(id):
    t = ET.Topic(generate_uid())
    a = FX.StartActor(input=t, initial_state={'history': [], 'system_prompt': '...'}, handler=chat) | run
    sessions[id] = (t, a)

# route a message:
def send(id, text):
    t, _ = sessions[id]
    FX.Publish(target=t, content=text) | run

introspect live state

# how long is this session's history?
FX.QueryActorState(actor=actor, transform=F.history | length) | run

# all running sessions
FX.QueryActorState() | run

reflect on how many layers are in play

count the zef ideas in use

✅ the | pipe
✅ ZefOps (parse_json, F.*, to_json, append)
✅ Zen notation (the dict state)
✅ Set-theoretic types (String, Any in signature)
✅ Entities (ET.Topic)
✅ UIDs (🍃-...)
✅ FX effects as data (FX.HTTPRequest, FX.Print)
✅ @zef_function
✅ Actors + topics
✅ collect vs run

Every single chapter, folded into 50 lines of code. That's the framework.

what this suggests for bigger systems

Think of every app as:

Topics for communication channels
Actors holding stateful logic
Databases/signals for persistence
HTTP server(s) exposing endpoints
Pure ZefOps for data transformations
FX values for effects at the boundary

Wire them up with pipes and topics. Observe them with FX.Query*. Persist them to files with FX.Save*. The primitives are small; the compositions are large.

modify it

Pick one:

Make the agent cap history at the last 20 turns
Add a "summarize" command that asks the LLM to summarize the conversation
Persist history to a DictDatabase between runs
Add a second actor that streams partial responses

Next up: the zen of zef — the 10-line manifesto. →