Introduction I was using AI agentic tools for work and hobby projects for last two years. Sometimes achieving results I couldn’t dream of before. And sometimes just getting unmaintainable code that smells. I was using most of major models and their ecosystems, but lately I tried to run OpenCode and Pi over the local models. The last one performed surprisingly well, thanks to it’s minimal feature set. That’s how I understood — it’s finally a time when some relatively small model running on your laptop can become really useful. But what I really wanted is to get full control and for that I need to build an agent myself. I have very strong believe that code around the LLM can sometimes bring even more progress in terms of agentic capabilities than new bigger, faster and smarter LLM release. While agentic tools when being used are often feel like a magic, I want to show it’s not. And I’ll show it by building my own agent step by step. In this article we’ll talk about the very core concept — “What should happen when you drop a message into the chat?” Defining the problem For now we need to put aside sophisticated memory management and even the tools which actually give our LLM agentic capabilities and focus on very base function: When I type something to the Agent, it should respond back. Look at the next diagram: The “AI” part is where the magic lives, so let’s try to make it a bit more boring and familiar. Almost same process, right? Some data gets through slow and not very predictable element that produces another data. What I’m trying to say — AI problems are not that different from other business problems that generations of people are most likely solved. We just need to dig a bit into the history to find what pattern suits the current need well. Getting hands dirty You can find full code on GitHub As we saw this basic message processing is basically a very heart of any AI system, if you ask an LLM to build some agent you’ll probably get a very naive solution. Some variation of the next code. Naive solution Honestly I’m not going to try getting it work and provided this block just for the reference const history = [{ role: "user", content: "hello" }]; let thinkingBuffer = []; let messageBuffer = []; let isBusy = false; // Poll new messages every 100ms setInterval(() => { // Prevent parallel execution if (isBusy) return; // take last const lastMessage = history.at(-1); // React on user message only if (lastMessage?.role === "user") { // clean old streams thinkingBuffer = []; messageBuffer = []; // compose chat context const context = buildContext(history); const modelStream = callModel(history); const message = ""; const thinking = ""; // process tokens from llm for (part of await modelStream) { // we want to handle thinking differently if (!part.message.thinking) { thinkingBuffer.push(part.message.thinking); } if (!part.message.content) { messageBuffer.push(part.message.content); } } // return AI message history.push(message); } }, 100); For sure, you can take a piece of code like that and make it work. But what’s the price? Except the expensive LLM tokens and time lost to make everything work, there’s another issue: Every time there’s a need for a new feature or an edge case found, you will end up with another condition, progressively making your code less readable and expensive to maintain. In my opinion, that’s currently the biggest problem with all agent generated code. Introducing Reactive Programming Returning to our call center analogy, we can think of almost any agent as a Stream of Events that shaping the data coming through. This mental model have it’s own programming style called Reactive programming. The reactive paradigm is build on top of Functional Programming — another very powerful concept. While in FP everything is a Function , in RP everything is a Stream . There’s a plenty of great materials on both topics, here we’ll just focus on the next features that will make our Agent developer’s life a bit easier: Declarativity — imagine you’re doing a while loop just to poll for the new messages, once you wrote the code, if you’re sane person, you may want to put it aside, cover by tests and never look at this again, so you end up with something like: function checkForNewMessage() { while (true) { // Rest of your ugly logic } } Next time you read this code, you just see checkForNewMessage that gives you enough knowledge what is function for. No need to dig into it’s content. Predictability — calling function with same arguments will always produce same results checkForNewMessage(history) // Message | null For example, the process shouldn’t be destroyed if LLM returns an Error, it either should wrap Error into another Message or do nothing, letting User or other Actor to decide what’s the next move. Composition — New capabilities are added by extension, not by hardcoding more rules, compare for instance this imperative code: function processMessage() { if (message.type === "user") { // do operation X }; if (message.type === "agent") { // do operation Y }; } And this function processUserMessage() { if (message.type !== "user") { return; } // do operation X } function processAgentMessage() { if (message.type !== "agent") { return; } // do operation Y } const applyHandlers = (x) => [processUserMessage, processAgentMessage].map(fn => fn(x)) While first case can look less esoteric at the first glance, think what will happen if you need to add more handlers or even react on some event in several places of the application. For instance, you may want to show some loading, process response and call context compaction independently, while trigger all of then on the same event. Applying Rx to agents Rx.js is a battle-tested library to handle reactive streams . While there’re specialized frameworks to build agents I think they all are built around either memory management or integrations with as much providers as exist. Or often both. Turning them down in favor of domain agnostic library will give you a freedom to design a unique workflow that aligns with your own needs. To reimagine our process of watching for the user messages: You can have something like: let messages = [] let currentStream = [] while(true) { if(messages[0]?.role === "user") { const context = messages.join("\n") const llmStream = await callLlm(context); for (part of await llmStream) { currentStream.push(part.message.content) } const message = currentStream.join("") messages.push(message) } To represent it as a reactive stream, we need to turn this into requirements: Trigger when new message appears Ignore any messages that are not coming from user Accumulate model’s stream tokens into new message Post message at the end We don’t want to poll for new messages, instead we want to be notified when message appears. Rx provides a class called Observable , which will do exactly that, let’s abstract from Agents for a while and think of processing text you type into the terminal: fromEvent(process.stdin, "line") // Create an observable from Node’s input .subscribe({ next: (line) => console.log(line) // when line appears log it back to terminal }) Now imagine, we want to apply some transformations to the data during the stream, we can do that using the operators, such as filter and map along with many other: fromEvent(process.stdin, "line") // Create an observable from Node’s input .pipe( filter(line => line.test(/Hello/), map(line => line.toUpperCase()), ) .subscribe({ next: (line) => console.log(line) // when line appears log it back to terminal }) In this code snippet we’ll accept only messages containing “Hello” and capitalize them before echoing back into the terminal. The part created with .subscribe block is called Observer , it provides callbacks where you can plug your handlers. But for our history processing we need something that can be updated from outside and at the same time will notify the subscribers when being updated. So for exactly this use case there’s a Subject exists const lines$ = new Subject() fromEvent(process.stdin, line) // Create an observable from Node’s input .subscribe({ next: (line) => lines$.next(line) // Update subject }) lines$.subscribe(console.log) // Log lines when appear on the subject fromEvent(process.stdin, line) // Create an observable from Node's input .subscribe({ next: (line) => lines$.next(line) // Update subject }) lines$.subscribe(console.log) // Log lines when appear on the subject With this basic knowledge we’re ready to build our history state The history state management As we discussed earlier we want agent to react to a human message, so our state can be represented by plain array. Rx will help us to get a fresh snapshot of it every time something changes. const initialState = { history: [] }; const messageAction$ = new Subject(); export const postMessage = (message) => messageAction$.next({ ...message, id: Math.random().toString(36).slice(2, 9), created: new Date() }); export const state$ = messageAction$.pipe( // On each event, notify subscribers with new copy of history state scan( (state, message) => ({ ...state, history: [...state.history, message] }), initialState ), shareReplay(1) ); Looks suspiciously familiar isn’t it? Well that’s because it’s a Reducer — the pattern behind Redux , which once was a most popular state management library in React. In fact you can go ahead and use Redux here along with Rx, to get maximum of both. Since state is pretty simple, I’ll just stick with less libraries. What happens in the code?: When postMessage is called, it emits new event into messageAction stream The scan operator is similar to reduce , each event is being accumulated. The difference with reduce is that every update is being emitted down the stream. The shareReplay(1) ensures all subscribers will get last update. In other words we don’t need to know the history of state changes, but we want very last snapshot of it to be broadcasted to all subscribers. Handling the LLM While in our naive solution streaming the LLM was just another loop, now we’d like to handle it in the right place, by introducing a dedicated observable that encapsulates all of the transformations and produces result we expect. I will use Ollama that allows you to run LLMs locally. If you don’t want or cannot run local models you’d probably like to obtain some free API key to play with, I think you’ll figure it out yourself. Caution! Local models are heavy and require a lot of RAM, you may want to opt into lighter 4b or 8b model const OLLAMA_MODEL = "gemma4:26b"; const transformResponsePart = (part) => { const thinking = part?.message?.thinking || ""; const content = part.message.content || ""; return { thinking, content }; }; export const fetchOllama$ = (context) => // Convert async generator to an observable defer(() => from( ollama.chat({ model: OLLAMA_MODEL, messages: [{ role: "user", content: context }], stream: true }) ) ).pipe( // Stream each token mergeMap((res) => from(res)), // return { thinking, content }; map(transformResponsePart) ); Shortly speaking what happens here is that we’re wrapping the Promise to an observable, so we can smoothly process it’s stream. As I said — feel free to swap Ollama with any other API call of your favorite model! This doesn’t matter for this codebase to work, you just probably need to slightly align mapping to stream tokens following the same contract. Now let’s see what’s each operator does: defer — creates an Observable only when someone subscribes to it. Ensuring our model is not being called before anybody needs it. from — converts promise into an Observable, so we can apply Rx operators on it’s stream. mergeMap takes all of the events LLM streams back and merges them into the main stream. map — applies helper function to transform the result Handle model stream The model response is actually not just a Promise, but async iterator, that emits tokens while model streams output tokens. For smooth user experience and observability we want to render both, so I’ll introduce two more subjects // Streaming state const thinkingSubject$ = new Subject(); const writingSubject$ = new Subject(); export const cleanupStreamTokens = () => { thinkingSubject$.next(""); writingSubject$.next(""); }; export const streamResponseTokens = (res) => { thinkingSubject$.next(res.thinking); writingSubject$.next(res.content); }; Assembling our Agent together export const buildContext = (history) => { return [...history.map((m) => ` ${m.role.toUpperCase()}: ${m.content}`)].join( "\n\n" ); }; state$ // No polling needed new message will trigger the pipe .pipe( Rx.filter((state) => { // React on Human messages only const lastMessage = state.history.at(-1); return lastMessage?.role === 'user }), // reset previous stream Rx.tap(cleanupStreamTokens), // Compose context for the model Rx.map((state) => buildContext(state.history)), // Process one message at the time, sequentially Rx.concatMap((context) => { return fetchOllama$(context).pipe( // Stream thinking and response tokens Rx.tap(streamResponseTokens), // Accumulate tokens Rx.reduce((acc, res) => acc + res.content, ""), // In the end, update history with a new message Rx.tap((finalMessage) => { postMessage({ role: "agent", content: finalMessage }); }) ); }) ) .subscribe(); That’s basically it! We built a very core of our agent, that can response user messages in conversational manner. Note that now we use reduce to accumulate the final agent response, since we want it to be emitted only once, when LLM is done streaming. Another thing worth attention is that the tap operator, managing all side effects is what gives you almost unlimited possibilities. For instance, if you want to monitor your agent — you just do tap(logger.info) in all places you want to. No need for additional tool. Connecting the interface Now we’re ready to connect an interface, if you want to build a nice feature reach web app or TUI, please go ahead, but I deliberately want to keep it minimal. Once again we’ll start with requirements: User should be able to send message By typing and hitting “Enter” By pasting the text from clipboard 2. For better observability and user experience we want to stream model thinking and response tokens in the real time. 3. Indicate loading or transitional state, e. g. “Thinking…”. I want to avoid spinners for now, because they’ll most likely require tracking cursor position and this is too much of a rabbit hole. Here I must introduce another concept that’s important for building an Agent, the turn . In our current system that’s basically everything what is happening between the User message and the Agent response. Particularly we’re interested in determining when we should start rendering agent stream and when to show that it’s done. So let’s create a helper to react on last message of provided role export function onMessage$(role) { return state$.pipe( map((s) => s.history.at(-1)), distinctUntilChanged((a, b) => a?.id === b?.id), filter((m) => m?.role === role) ); } Now our boundaries will be as simple as const turnStart$ = onMessage$("user"); const turnEnd$ = onMessage$("agent"); Finally, we can process our thinking and message streams turnStart$ .pipe( switchMap(() => thinkingSubject$.pipe( takeUntil( writingSubject$.pipe( filter((t) => !!t), take(1) ) ), tap((tokens) => process.stdout.write(`\x1b[37m${tokens}`)), finalize(() => { process.stdout.write("\n\n\x1b[38;5;255mAgent is typing...\n"); }) ) ) ) .subscribe(); turnStart$ .pipe( switchMap(() => writingSubject$.pipe( takeUntil(turnEnd$), tap((tokens) => process.stdout.write(`\x1b[38;5;12m${tokens}`)), finalize(() => { process.stdout.write(`\n\n`); rl.setPrompt("\x1b[38;5;46m>>> \x1b[38;5;255m"); rl.prompt(); }) ) ) ) .subscribe(); Note: You may notice that I’m using raw escape codes here, for example x1b[38;5;12m – this is just a low level way to handle the colors without using extra libraries. Escape codes are not that hard, if you’re interested you can read about them here https://notes.burke.libbey.me/ansi-escape-codes/ Few new things worth explanation here: switchMap — cancels parent stream, as we only need to know that turn is started and not interested in it’s data takeUntil — completes the stream when other observable passed into it, emits an event finalize — Fire side effect after stream completes — I want to add some text and lines between messages For the last touch, we need to prompt user for the input, I’ll use a simple Node.js readline interface: import readline from "node:readline/promises"; const rl = readline.createInterface({ input: process.stdin, output: process.stdout }); rl.setPrompt(">>>"); rl.prompt() Now we need to handle the readline events, that’s another time when Rx comes very handy const input$ = fromEvent(rl, "line"); const done$ = input$.pipe( bufferTime(500), // Handle paste filter((lines) => lines.at(-1)?.trim() === "") // handle double RETURN ); defer(() => { rl.setPrompt("\x1b[38;5;46m>>> \x1b[38;5;255m"); rl.prompt(); return input$.pipe( tap(() => { process.stdout.write("\x1b[37m...\n\x1b[38;5;255m"); }), buffer(done$), map((msg: string[]) => msg.join("\n")), tap((content: string) => { postMessage({ role: "user", content }); process.stdout.write("\x1b[38;5;255mAgent is thinking...\n"); }) ); }).subscribe(); New things here: buffer and bufferTime , they will accumulate the events and emit them all together when condition is met. Since I want to keep code simple, the submit will happen after two consecutive empty lines (You’ll need to hit enter twice). But I don’t want pasted text to immediately trigger submit, so we add short delay to collect all pasted lines with bufferTime. Running the agent Decide what is your entry point and run node index.js Or if you use typescript npx tsx index.ts See full version GitHub Conclusion We just built a base for an agent in about 170 lines of code! While it lacks many features that are required for production ready system, such as observability, error handling, proper context compaction, etc, I want to assure you that adding these new capabilities is very easy using reactive programming principles. I hope now you understand the principles better and able to start building your own agentic systems! The next article is going to cover the tools. Thanks for reading! Subscribe for free to receive new posts and support my work. I’d appreciate if you provide me any feedback or share this work with your friends! Acknowledgements Huge thanks to my wife Yulia for mentoring me on Rx, architecture principles as well as reviewing my code and this article. During our long weekend walks we discussed how pipelines should work and Yulia even drew architecture diagrams while we were sitting in coffeeshops. Without her being truly involved in my learning I’d never have achieved this depth of understanding. Originally published at https://vasilymitronov.substack.com . Build a reactive AI agent harness — Part 1. Conversation. was originally published in GoPenAI on Medium, where people are continuing the conversation by highlighting and responding to this story.