Program A · AI engineering · ThoughtFlow
Cognition in ordinary code.
Most agent frameworks sell top-level simplicity by hiding the loop. ThoughtFlow does the opposite: fourteen small primitives, a memory object that is the whole state, and a loop you write yourself. It looks like more code. It is less system.
The engineer who will be accountable for the behavior should be the one who wrote the control flow.
Hidden runtimes are pleasant on day one. On day forty, when a customer asks why the agent did something, the answer is buried inside a scheduler you cannot step through. ThoughtFlow keeps every decision point in code you own. The trade is simple: you write the loop. In exchange, there is no behavior in the system that you did not author.
The same pattern played out once before. Multilayer perceptrons became useful when engineers could see and shape every layer; the frameworks that abstracted the architecture away did not survive contact with hard problems. We expect agent engineering to rhyme.
the loop is yours
# Every primitive has the same shape: memory in, memory out.
# So the control flow is plain Python, and it is yours.
while not memory.get_var("done"):
memory = plan(memory) # THOUGHT
memory = act(memory) # ACTION
memory = reflect(memory) # THOUGHT
memory = check_done(memory) # DECIDE
# What happened? Every step is in the log, in order.
for event in memory.get_events():
print(event["type"], event["content"])
LLMOne client, seven providers, structured output.
MEMORYEvent-sourced state. The whole system state.
THOUGHTOne LLM call with a named result.
TOOLA function with a contract the model can read.
ACTIONExecute a tool call; record what came back.
AGENTA loop you wrote, packaged.
DECIDERoute between named options.
PLANProduce a list of steps as data.
WORKFLOWFixed sequence of steps.
DELEGATEHand memory to another agent.
CHRONTime and schedule as primitives.
CHATConversation over memory.
EMBEDVectors with the same client shape.
MCPModel Context Protocol client.
test_refund_policy.py
from thoughtflow import LLM, MEMORY
from thoughtflow.eval import Harness, TestCase
# Record once against a live model...
recorded = MEMORY()
live = LLM("openai:gpt-4o").record(recorded)
support_flow(MEMORY(), live)
recorded.save("fixtures/refund.json")
# ...then replay forever: offline, no keys, same decisions.
replay = LLM.replay(MEMORY.from_json("fixtures/refund.json"))
def escalates(memory):
return memory.get_var("route_result") == "escalate"
case = TestCase(name="broken item above authority",
messages=[{"role": "user",
"content": "Refund order 8812, it arrived broken."}],
check=escalates)
results = Harness([case]).run(lambda m: support_flow(m, replay))
assert results.passed_count == results.total_count
Record a run once. Replay it forever.
Models do not give the same answer twice, so tests that check the answer are flaky by design. ThoughtFlow tests check the decision instead: which branch was taken, which tool was refused, which commitment was kept. Record one run against a live model, save the memory, and replay it as a fixture from then on. The suite runs in milliseconds, offline, with no API key.
This is why memory is a log. A log can be replayed, and a replay can be tested. Anything less means trusting the model to behave the same way tomorrow.
Releases17
Providers7
Import time~15 ms
Deps0