Core integration
The class that builds its own engines cannot be run without them
Three volumes produced three working parts: something that hears you, something that thinks, something that answers in her voice. Chapter 10 ran them in one loop, and chapter 31 gave every one of them a single place to keep its model name and its device. What is still missing is the object in the middle. Top-level functions are fine for one script; a system with a health check, a scheduler, a command dispatcher and a metrics log all wanting to drive a conversation needs one class that owns a turn.
The obvious way to write that class is to let it build what it needs. One line in the constructor for Whisper, one for Ollama, one for F5-TTS, and the class is self-sufficient. It works exactly once, on the machine where all three are installed, with several gigabytes of weights loaded before the first line of your own logic runs. Then you try to check a small rule (does an empty transcript really skip the model call?) and discover there is no way to ask without paying that cost again. Swapping Whisper for a smaller model means editing the class. Running headless on a machine with no audio output means editing the class. Every experiment routes through a constructor you did not want to touch.
So the rule for this chapter, and for every composed object after it: the core receives its collaborators as arguments and never constructs them. The caller decides what to hand in. Production hands in the real engines. A test hands in three functions that return fixed strings and finish in microseconds. Nothing hands in nothing, and the core still runs.
Other languages express "something that transcribes" as an interface, and Python
offers two imitations: inherit from an abstract base class, or declare a
Protocol. Both make a real transcriber a class with a required method
name, which then makes a fake transcriber a class too. Since every collaborator here
is one call in and one value out, the alias
Callable[[str], str] says everything the core needs to know and leaves
a plain function, a bound method, and a lambda all equally acceptable.
Type aliases document the contract for readers and for the type checker; they
constrain nobody at runtime.
Four methods, each one safe when its collaborator is missing
# glados/core.py
from typing import Callable, Optional
Transcriber = Callable[[str], str]
Thinker = Callable[[str, list[dict]], str]
class GladOSCore:
def __init__(self, stt: Optional[Transcriber] = None,
llm: Optional[Thinker] = None) -> None:
self.stt = stt
self.llm = llm
self.history: list[dict] = []
def process(self, audio_path: str) -> str:
if self.stt is None:
return ""
return self.stt(audio_path)
def respond(self, text: str) -> str:
if self.llm is None:
return f"[GLaDOS] Received: {text}"
return self.llm(text, self.history)
if __name__ == "__main__":
core = GladOSCore()
print(repr(core.process("glados/data/heard.wav")))
print(core.respond("hello"))
$ uv run python glados/core.py
''
[GLaDOS] Received: hello
The first version of the core is deliberately the emptiest one: no arguments passed,
no models installed, and it still produces output. process returning
"" when there is no listener is a promise the rest of the loop will
lean on, because "no listener" and "five seconds of room tone" should behave
identically downstream. The repr() in the demo lets you see the empty
string instead of a blank line.
Speaker = Callable[[str], None]
class GladOSCore:
def __init__(self, stt: Optional[Transcriber] = None,
llm: Optional[Thinker] = None,
tts: Optional[Speaker] = None) -> None:
self.stt = stt
self.llm = llm
self.tts = tts
self.history: list[dict] = []
def speak(self, text: str) -> None:
if self.tts is None:
print(f"[TTS] {text}")
return
self.tts(text)
if __name__ == "__main__":
GladOSCore().speak("No voice engine wired in yet.")
$ uv run python glados/core.py
[TTS] No voice engine wired in yet.
Each fallback answers a design question about what absence should mean. For a
missing listener, silence. For a missing brain, an echo you can recognise as a
placeholder. For a missing voice, a printed line, because raising here would make
the loop unrunnable on any machine without F5-TTS loaded, and that is most machines
most of the time. The prefix [TTS] keeps a headless transcript readable
beside a spoken one.
MAX_TURNS = 20
def run_turn(self, audio_path: str) -> bool:
text = self.process(audio_path)
if not text.strip():
return False
print(f"You: {text}")
reply = self.respond(text)
print(f"GLaDOS: {reply}")
self.speak(reply)
self.history.append({"role": "user", "content": text})
self.history.append({"role": "assistant", "content": reply})
self.history = self.history[-2 * MAX_TURNS:]
return True
if __name__ == "__main__":
def fake_stt(path: str) -> str:
return "Hello GLaDOS"
def fake_llm(text: str, history: list[dict]) -> str:
return f"Oh. It's you again. ({len(history)} messages remembered)"
def fake_tts(text: str) -> None:
print(f"[SPEAKING] {text}")
core = GladOSCore(stt=fake_stt, llm=fake_llm, tts=fake_tts)
core.run_turn("glados/data/heard.wav")
core.run_turn("glados/data/heard.wav")
print(f"history holds {len(core.history)} messages")
$ uv run python glados/core.py
You: Hello GLaDOS
GLaDOS: Oh. It's you again. (0 messages remembered)
[SPEAKING] Oh. It's you again. (0 messages remembered)
You: Hello GLaDOS
GLaDOS: Oh. It's you again. (2 messages remembered)
[SPEAKING] Oh. It's you again. (2 messages remembered)
history holds 4 messages
A complete conversational turn just ran twice with no model on the machine, in the
time it took Python to start. Three details carry weight. The empty-transcript guard
returns False instead of None, so a caller can count real
turns without inspecting anything else. History stores one message per speaker in
the role-and-content form Ollama expects, so it can be passed straight through to a
chat call with no translation at the boundary. And the slice from chapter 29 caps
it: forty messages, twenty exchanges, evicted oldest-first by the same one-line
trick that bounded the sensor log. Nothing here says the word Whisper, and the fake
brain proves history reaches the brain by counting it.
# labs/wire_core.py
import json
from pathlib import Path
import ollama
import sounddevice as sd
import soundfile as sf
from f5_tts.api import F5TTS
from faster_whisper import WhisperModel
from glados.core import GladOSCore
from labs.system_config import SystemConfig, build_default_config
from labs.glados_voice import REF_FILE, REF_TEXT
AUDIO_OUT = "glados/data/glados_says.wav"
def build_core(config: SystemConfig) -> GladOSCore:
stt_cfg = config.get_component("stt").settings
llm_cfg = config.get_component("llm").settings
tts_cfg = config.get_component("tts").settings
persona = json.loads(Path("configs/personality.json").read_text())
print(f"Loading whisper {stt_cfg['model']} on {stt_cfg['device']}...")
whisper = WhisperModel(stt_cfg["model"], device=stt_cfg["device"],
compute_type=stt_cfg["compute_type"])
print("Loading F5-TTS...")
tts = F5TTS(device=tts_cfg.get("device"))
def transcribe(path: str) -> str:
segments, _ = whisper.transcribe(path, language="en")
return " ".join(s.text.strip() for s in segments)
def think(text: str, history: list[dict]) -> str:
messages = [{"role": "system", "content": persona["system_prompt"]}]
messages += history + [{"role": "user", "content": text}]
response = ollama.chat(model=llm_cfg["model"], messages=messages)
return response["message"]["content"]
def synthesize(text: str) -> None:
tts.infer(ref_file=REF_FILE, ref_text=REF_TEXT, gen_text=text,
file_wave=AUDIO_OUT, remove_silence=True)
data, sample_rate = sf.read(AUDIO_OUT)
sd.play(data, sample_rate)
sd.wait()
return GladOSCore(stt=transcribe, llm=think, tts=synthesize)
if __name__ == "__main__":
core = build_core(build_default_config())
core.run_turn("glados/data/heard.wav")
$ uv run python labs/wire_core.py
Loading whisper base on cpu...
Loading F5-TTS...
You: Are you awake in there?
GLaDOS: I was. Then you spoke.
Every import that costs a gigabyte lives in this file, and none of them live in
glados/core.py. build_core is the composition root: the one
place that knows which concrete engines exist, reads their settings out of the
registry, constructs them once, and closes over them in three small functions with
the signatures the core expects. Your reply text will differ from the one above, and
so will your timings; the wiring will not. Note what the closures buy you.
transcribe captures whisper, so the model loads once even
though the function runs every turn, and the core never learns a model was involved.
Why this works: the core knows a call signature, not a class
GladOSCore contains no mention of Whisper, Ollama or F5-TTS. What it
knows about self.stt is one sentence: call it with a path string, get a
text string back. Everything that satisfies that sentence is interchangeable, and
three very different things do. A closure over a loaded model is one. A one-line
function returning "Hello GLaDOS" is another. None is the
third, handled by a guard instead of a crash. All three flow through the identical
run_turn, because the method calls a name, and Python resolves that name
at the moment of the call.
The general move is one you will make repeatedly for the rest of the book: construction and use are separate jobs, and they belong in separate places. Use lives in the core, where the interesting logic is and where you want to run experiments cheaply. Construction lives at the edge, in one function that runs once at startup. Push construction outward and the middle of your system becomes testable, swappable, and describable in a paragraph. Chapter 33 depends directly on this split: you can only probe a component at startup if something outside the core is holding the component.
The None guards look like defensive clutter until the first time you
wire a partial core. Suppose you delete the guard in respond while
tidying, on the reasoning that an assistant always has a brain, then run a
listening-only script that wires the transcriber and nothing else:
def respond(self, text: str) -> str:
return self.llm(text, self.history) # guard deleted
# labs/hearing_test.py -- checking transcription only
core = GladOSCore(stt=transcribe)
core.run_turn("glados/data/heard.wav")
$ uv run python labs/hearing_test.py
You: Are you awake in there?
Traceback (most recent call last):
File "/home/you/GladOS/labs/hearing_test.py", line 12, in <module>
core.run_turn("glados/data/heard.wav")
File "/home/you/GladOS/glados/core.py", line 41, in run_turn
reply = self.respond(text)
^^^^^^^^^^^^^^^^^^
File "/home/you/GladOS/glados/core.py", line 27, in respond
return self.llm(text, self.history)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: 'NoneType' object is not callable
Read the message literally and it is precise: the object named self.llm
is None, and you tried to call it. The confusing part is that the
You: line printed before the crash, suggesting a healthy pipeline. It
was healthy up to the second collaborator. Python 3.11's caret markers point at the
exact call, and the fix is to restore the guard. The lesson generalises past this
one method: a default of None
is a promise that the attribute is optional, and that promise is kept in every
method that touches it or in none of them. A half-guarded core is worse than an
unguarded one, because it fails only on the paths you did not test.
Checkpoint, and a core nobody has checked yet
- I can state the three ways
self.sttcan be satisfied (a closure over a real model, a fixed-string function,None) and whatrun_turndoes in each case. - I can explain why
build_coreholds every heavy import andglados/core.pyholds none. - I know what each fallback means as a design decision: empty string for no listener, echo for no brain, printed line for no voice.
- I can read
TypeError: 'NoneType' object is not callablestraight to its cause, and name the rule that prevents it. - I can say why history is stored as role-and-content messages and why the
twenty-turn slice sits inside
run_turninstead of at the model call.
Exercise 1 — a command that beats the brain to the answer.
Add reset_history() and make respond short-circuit on the
phrase "forget everything" before the LLM is ever called. Verify with two turns and
a length print.
The short-circuit goes at the top of respond, above the
None check, so it works with or without a brain wired in:
def reset_history(self) -> None:
self.history = []
def respond(self, text: str) -> str:
if "forget everything" in text.lower():
self.reset_history()
return "History purged. How refreshing."
if self.llm is None:
return f"[GLaDOS] Received: {text}"
return self.llm(text, self.history)
One wrinkle to notice when you run it: run_turn appends the purge
exchange to the history it just emptied, so the length prints as
2, not 0. Whether that is a bug depends on what you
meant by "forget", and deciding is the actual exercise.
Exercise 2 — tee every transcript to a file. Without
editing GladOSCore, make every line she hears land in
glados/data/transcripts.log as well as reaching the brain.
Wrap the injected callable in another callable of the same signature, then inject the wrapper:
def tee(inner: Transcriber, path: str) -> Transcriber:
def wrapped(audio_path: str) -> str:
text = inner(audio_path)
with open(path, "a") as f:
f.write(text + "\n")
return text
return wrapped
core = GladOSCore(stt=tee(transcribe, "glados/data/transcripts.log"))
Tail the file while you talk and you will see it fill. The core cannot tell the difference, because the wrapper honours the same contract. New behaviour arrived by composition at the wiring site, with the orchestrating class untouched.
Exercise 3 — time the turn from the outside. Use chapter
28's @timed decorator to find which stage dominates a real turn,
again without editing the core.
Decorate the three closures in build_core and run one turn against
real models. Expect the ordering you measured in chapter 10 to hold: synthesis
far ahead of everything, transcription a second or two, the 3B model around a
second, with your figures depending on your CPU. You instrumented a running
system by changing only the file that assembles it. Keep the decorated version;
chapter 36 turns these numbers into a rolling metrics log.
The core composes, and with fakes it demonstrably runs a turn end to end. What nobody has established is whether the real collaborators work. A callable can be present, correctly typed, and still throw the first time it sees audio, because a model file moved or an Ollama daemon is down. Discovering that mid-conversation is the worst possible timing. Next chapter probes every wired component before the loop takes control, and reports the whole system's readiness in one object.