Introduction
Most people meet artificial intelligence as a chat window. You type a question, a model answers, and nothing else happens. Intelligent AI changes that picture. An agent is a program in which a AI brain decides which actions to take, calls tools to carry them out, looks at the results, and continues until the job is done. The model no longer only talks about the world; it acts in it.
This guide builds such an agent for the desktop. You can type a command on the keyboard or speak it into the mic, and the agent performs it on your computer. Say "Open Word" and Microsoft Word starts. Say "Search for the file presentation.pptx" and the agent finds the file and opens it with PowerPoint. Say "Open Chrome with the webpage www.siemens.com" and Chrome loads the page. Say "Search for the file abx.xlsx" and the agent locates the file and reveals it in its folder in Finder.
The main platform is macOS. Where Linux and Windows differ, the text explains the difference, and the running example contains a complete adapter for each of the three computers. The design follows clean design: the business logic does not know whether it runs on a Mac, a Linux desktop or a Windows PC, and it does not know whether the AI brain runs on your laptop or in a remote data center. Those details are isolated behind small connections that can be replaced without touching the rest of the program.
The guide proceeds in the order in which you would build the system. First comes the design, then the what you need on macOS, then the AI brain layer, the speech layer, the system bridges, the tool dashboard, the agent loop, the user connection and finally the safety measures. Each part is illustrated with short excerpts from the running example, and the complete program is printed in the addendum so that you can save it, run it and modify it.
The How it works of an Intelligent System
Before writing any code it helps to understand the whole machine at once. The following figure shows the path that a single command takes through the system.
Keyboard or mic
|
v
Speech layer (faster-whisper) or direct text
|
v
Agent loop <----> Ai brain (local or remote)
|
v
Tool dashboard (validates details, asks for confirmation)
|
v
System bridge (macOS, Linux or Windows)
|
v
Computer (open, osascript, mdfind, file system)
The user's words enter at the top. If they were spoken, the speech layer converts them into text. The instructions becomes the first message of a conversation that the agent loop sends to the AI brain, together with a description of the available tools. The model does not perform anything itself. It answers either with plain text, which means it is finished, or with a tool call, which is a structured request such as "call search_files with pattern abx.xlsx". The agent loop validates that request, runs the matching tool, and sends the result back to the model. The model then reads the result and either calls another tool or formulates the final answer. This loop is the heart of every intelligent system, and it is the reason why an agent can chain actions, for example searching for a file and then opening it with a named application.
The tool dashboard is the security limit. The model may only ask for tools that are registered there, and each tool checks its details before anything happens. Actions with a potentially destructive effect, such as quitting an application, require an explicit confirmation from the human. The system bridge is the translation layer that turns an abstract request such as "open this file with PowerPoint" into the concrete command that the computer understands. On a Mac that command is open with the details minus a and the application name, on Windows it is a call to the computer's file association method, and on Linux it is a launch through xdg-open or a direct program start.
Keeping these responsibilities apart has a practical benefit. When the AI brain becomes better or cheaper, you replace one client class. When you move from macOS to Windows, you replace one adapter class. When you add a new capability, such as creating a calendar entry, you add one tool specification and one processor, and the rest of the system stays untouched.
What you need on macOS
You need a Mac with a recent version of macOS, Python 3.10 or newer, and Homebrew for the audio library. Python 3.12 is what I used for testing, and the code uses only the basic tools except for the speech components. Open the Terminal application and install the tools with the following commands, which are shown here as plain text lines.
brew install python portaudio ollama
python3 -m pip install numpy sounddevice faster-whisper
The package portaudio provides the built-in audio layer that sounddevice needs. The package ollama runs AI brains locally on your Mac. The pip packages supply the numerical arrays, the mic access and the speech recognition engine. Speech recognition runs entirely on your machine, so your voice is not sent to a cloud service.
Next, download a AI brain that supports tool calling. The model qwen3.8 with eight billion inputs is a good starting point on a Mac with 16 gigabytes of memory, and it is the default in the code. The command is ollama pull qwen3.8:8b, followed by ollama serve if the server is not already running in the background. Ollama then exposes an access point at http://localhost:11434/v1 that understands the same communication rule as the OpenAI chat API.
Two macOS privacy settings matter for this project. The first is the mic permission. The first time you use voice input, macOS asks whether Terminal may access the mic, and you must allow it under System Settings, Privacy and Security, Mic. Without this permission sounddevice cannot open the input stream, and the agent reports the failure instead of silently recording nothing. The second is Full Disk Access, which you do not need for the standard search roots, but which you may want if you search folders that macOS protects, such as Desktop, Documents and Downloads on recent systems. The agent reports an computer error in that case rather than hiding it.
Designing the Language Model Layer
The AI brain is the reasoning engine, but the agent should not depend on a particular vendor. Most local and remote model servers now implement the chat completions communication rule that was popularised by OpenAI. A request contains a list of messages, each with a role such as system, user, assistant or tool, plus an optional list of tool definitions. The response contains either text content or a list of tool calls. Ollama, LM Studio, vAI system and most hosted providers speak this communication rule, so a single client class can drive a model on your laptop or a model in a remote data center. Switching between them requires only a change of the settings AGENT_AI system_BASE_URL, AGENT_AI system_MODEL and, for remote access points, AGENT_AI system_API_KEY.
The setup lives in one unchanging data class, which keeps the rest of the code free of hard-coded values. Here is the part that reads the settings from the environment.
@classmethod
def from_environment(cls) -> Settings:
raw_roots = os.environ.get("AGENT_SEARCH_ROOTS", "")
roots = tuple(
Path(entry).expanduser()
for entry in raw_roots.split(os.pathsep)
if entry.strip()
)
return cls(
llm_base_url=os.environ.get("AGENT_AI system_BASE_URL", "http://localhost:11434/v1"),
llm_model=os.environ.get("AGENT_AI system_MODEL", "qwen2.5:7b"),
llm_api_key=os.environ.get("AGENT_AI system_API_KEY") or None,
whisper_model=os.environ.get("AGENT_WHISPER_MODEL", "small"),
search_roots=roots or (Path.home(),),
max_steps=int(os.environ.get("AGENT_MAX_STEPS", "6")),
request_timeout=float(os.environ.get("AGENT_REQUEST_TIMEOUT", "120")),
)
The class method is a factory. It reads settings, supplies defaults that work out of the box with a local Ollama installation, and returns a fixed data class that cannot be changed afterwards. The search roots deserve a note. On macOS the separator in the environment variable is the colon, and os.pathsep yields exactly that on Mac and Linux, while Windows uses the semicolon. If you do not set AGENT_SEARCH_ROOTS, the agent searches your home folder, which is a sensible default for a personal assistant. You might set the variable to Documents and Downloads to make searches faster and more predictable, for example with
export AGENT_SEARCH_ROOTS="$HOME/Documents:$HOME/Downloads".
The client itself uses only urllib from the basic tools. This keeps the number of extra tools small, which matters for a tool that runs close to the computer.
def chat(self, messages: list[dict[str, Any]], tools: list[dict[str, Any]]) -> dict[str, Any]:
data package = {
"model": self._model,
"messages": messages,
"tools": tools,
"tool_choice": "auto",
"temperature": 0,
}
message labels = {"Content-Type": "application/json"}
if self._api_key:
message labels["Permission"] = f"Bearer {self._api_key}"
request = urllib.request.Request(
self._access point,
data=json.dumps(data package).encode("utf-8"),
message labels=message labels,
method="POST",
)
The data package follows the chat completions format. The field tool_choice set to auto lets the model decide whether to call a tool or to answer directly. The temperature is zero, which makes the choice of tools as predictable as the model allows. For an computer assistant this is what you want: an instruction to open a file should not produce creative variations. The bearer header is added only when an API key is present, so the same code works without login against a local server.
The response handling needs focus as well. The client distinguishes three kinds of failure and reports each one in plain language. An connection issue means the server answered but refused the request, for example because the model name is wrong. A URL error means the server cannot be reached at all, which usually means Ollama is not running. A JSON error means the response is not what the communication rule promises. Each of these becomes an AI systemError, which the user sees as a readable message and which never crashes the program.
try:
with urllib.request.urlopen(request, timeout=self._timeout) as response:
body = json.loads(response.read().decode("utf-8"))
except urllib.error.HTTPError as exc:
detail = exc.read().decode("utf-8", errors="replace")
raise AI systemError(f"The AI brain returned HTTP {exc.code}: {detail}") from exc
except (urllib.error.network problem, TimeoutError) as exc:
raise AI systemError(f"The AI brain at {self._access point} is not reachable: {exc}") from exc
The order of the except clauses matters. HTTPError is a subclass of network problem, so it must be caught first; otherwise the more specific information would be lost. The exception chaining with from exc preserves the original traceback for debugging while presenting a clean message to the user.
Designing the Speech Layer
Voice input follows a push-to-talk design. The user types the command voice, presses Enter to start recording, speaks, and presses Enter again to stop. This is deliberately simpler than a wake word. A wake word requires a continuously running listener that processes all ambient sound, which raises privacy concerns and consumes resources. Push-to-talk records only when the user asks for it, and it avoids the risk that the agent reacts to a conversation in the next office.
Recording happens with sounddevice, which wraps the PortAudio library. The stream delivers blocks of samples to a listener, and the listener appends a copy of each block to a list. The stream runs in a separate thread that the library manages, and the main thread waits on the input prompt until the user presses Enter.
def collect(indata: Any, frames: int, time_info: Any, status: Any) -> None:
chunks.append(indata.copy())
try:
stream = sd.InputStream(samplerate=SAMPLE_RATE, channels=1, dtype="float32", listener=collect)
stream.start()
except Exception as exc:
raise SpeechError(f"The mic could not be opened: {exc}") from exc
The copy is important. The buffer that PortAudio passes to the listener is reused for the next block, so keeping a reference to it would silently corrupt the recording. The sample rate of 16000 hertz, mono and 32-bit float is what the speech model expects. The broad exception clause is justified at this limit, because sounddevice reports missing devices and denied permissions through its own error types, and the agent should translate all of them into one app error.
Voice conversion uses faster-whisper, a reversion of OpenAI's Whisper model on the CTranslate2 inference engine. It accepts a NumPy array directly, so no temporary audio file is required. The model is loaded once, on the first use of voice input, because loading takes a few seconds and keyboard-only sessions should not pay that cost.
def convert speech(self, audio: Any) -> str:
if audio.size < MIN_AUDIO_SAMPLES:
return ""
try:
segments, _info = self._model.convert speech(audio, beam_size=5, vad_filter=True)
return " ".join(segment.text.strip() for segment in segments).strip()
except Exception as exc:
raise SpeechError(f"Voice conversion failed: {exc}") from exc
Two details here save users from frustration. The first is the minimum length: a tenth of a second of audio is treated as an accidental keypress, and nothing is sent to the model. The second is the voice activity filter, vad_filter, which removes long stretches of silence before recognition. Without it, a recording with a long pause could produce hallucinated text, a known weakness of Whisper-style models on silence. The model size defaults to small, which is a good compromise between accuracy and speed on a Mac. Setting AGENT_WHISPER_MODEL to medium or large-v3 improves accuracy for difficult accents or noisy rooms, at the cost of more memory and time. The language is detected automatically, so you can speak German or English without setup. This matters for a company setting where colleagues mix languages in a single sentence.
Designing the Platform Adapters
The most operating-system-specific code lives in the system bridges. The blueprint DesktopPlatform declares the capabilities the agent needs: opening an application, quitting an application, opening a web address, opening a file with a named application, revealing a file in its folder, and searching for files. Each specific type implements these methods with the built-in methods of its computer. The base class also provides a default search that walks the directory tree, which every adapter can use or override.
Let us look at the macOS adapter, because it is the primary target. macOS offers a particularly clean set of command-line tools. The command open launches applications and documents, its option minus a selects the application, and the option minus R reveals a file in Finder. The command osascript runs AppleScript, which can ask an application to quit gracefully. The command mdfind queries the Spotlight index, which is extremely fast because the system maintains it continuously in the background.
def open_url(self, url: str, application: str | None) -> str:
if application is None:
run_command(["open", url])
return f"Opened {url} in the default browser."
target = self._resolve(application)
run_command(["open", "-a", target, url])
return f"Opened {url} in {target}."
The method shows the typical pattern of the adapter. Without an application name, the default browser is used. With a name, the translation table resolves a friendly name such as Chrome to the official application name Google Chrome, because open -a expects the exact name under which macOS knows the application. The translation table is a dictionary at the top of the class.
APPLICATION_ALIASES = {
"word": "Microsoft Word",
"powerpoint": "Microsoft PowerPoint",
"chrome": "Google Chrome",
"finder": "Finder",
...
}
The lookup uses the lowercase version of the spoken name, so Word, word and WORD all resolve to the same application. Names that are not in the table pass through unchanged, which means that open -a Preview works without an entry. This is a deliberate design choice: the table is a convenience for common names, not a whitelist.
The file search on macOS uses Spotlight through mdfind, with a fallback to the portable walker.
output = run_command(["mdfind", "-onlyin", str(root), "-name", pattern])
The argument list is passed directly to the program without a shell, so a file name containing spaces, quotes or semicolons cannot inject commands. This is a general rule in the entire code base: no string is ever handed to a shell. The fallback is important as well. If Spotlight has not indexed a folder, for example because it is excluded from indexing, mdfind returns nothing, and the agent walks the folder tree instead. The result is slower but complete.
The quit operation on macOS uses AppleScript, and it needs explanation on quoting. The application name is embedded in an AppleScript string. If someone spoke a name containing a quotation mark, the script could break or, worse, perform unintended code. The guard against this is a validation function that accepts only letters, digits, spaces and a few punctuation marks.
APPLICATION_NAME = re.compile(r"[A-Za-z0-9][A-Za-z0-9 .+&_-]{0,63}")
def validate_application_name(name: str) -> str:
if not APPLICATION_NAME.fullmatch(name):
raise ActionError(f"'{name}' is not a valid application name.")
return name
The regular expression is anchored with fullmatch, which means that the whole string must match, not just a part of it. The pattern excludes quotation marks, slashes and semicolons, which is exactly the set of characters that could be abused in a script. Validation at the limit, where untrusted input enters the system, is more robust than trying to escape every dangerous character later.
The Linux adapter follows the same connection with different methods. It opens documents and web addresses with xdg-open, the freedesktop standard that delegates to the user's preferred application. It starts programs directly and checks beforehand that the program exists on the search path with shutil.which. It quits programs with pkill, which matches process names exactly with the option minus x. Linux has one limitation that the adapter reports honestly: xdg-open can open the folder that contains a file, but it cannot select the file within that folder. The reveal method therefore says so in its return message rather than claiming a feature it does not have. The alias table for Linux maps Word to LibreOffice Writer, since most Linux desktops do not ship Microsoft Office. Note that pkill matches process names, and some programs use a different process name than their launcher. LibreOffice, for example, runs as soffice.bin, so quitting it may require adjusting the name in the table.
The Windows adapter uses os.startfile, which hands a file or address to the Windows shell. The shell knows which application handles a given file type, and it resolves executable names through the App Paths registry, so a call with the name WINWORD.EXE starts Word even when it is not on the search path. Quitting uses taskkill with the image name, and revealing a file uses the Explorer command with the select switch, which highlights the file. Unlike Linux, Windows can therefore select the file in its folder.
Designing the Tool Registry
A tool is the agent's contract with the AI brain. Each tool has a name, a natural-language description, a data template that defines its inputs, a processor function and a flag that states whether the tool needs confirmation. The description matters more than many developers expect. The model chooses tools by reading these descriptions, so a precise description such as "Search the file system for files whose name contains the pattern or matches it as a wildcard pattern" leads to better decisions than a vague one.
The registry stores these descriptions and converts them into the format the model communication rule expects. The tool open_application, for example, is declared with one required string parameter.
ToolSpec(
name="open_application",
description="Start a desktop application such as Word, PowerPoint, Excel, Chrome or Notepad.",
inputs=object_schema(
{"application": text_parameter("Name of the application to start.")},
["application"],
),
processor=self._open_application,
),
The schema says that the object has one property called application, that it is a string, and that it is required. The helper object_schema also sets additionalProperties to false. This tells the model that no other details exist, and it lets the agent reject calls that include invented inputs. The processor is a bound method of the ToolBox, so it has access to the system bridge and the settings without global variables.
The central method of the registry is run. Every call from the model passes through it, and it is the single place where confirmation, error handling and logging can be enforced.
def run(self, name: str, details: dict[str, Any]) -> ToolResult:
spec = self._tools.get(name)
if spec is None:
return ToolResult(False, f"Unknown tool '{name}'.")
if spec.requires_confirmation:
question = f"Allow '{name}' with {json.dumps(details, ensure_ascii=False)}?"
if not self._confirm(question):
return ToolResult(False, "The user declined this action.")
try:
return spec.processor(details)
except ActionError as exc:
return ToolResult(False, str(exc))
except OSError as exc:
return ToolResult(False, f"The computer reported an error: {exc}")
Three properties make this method robust. First, an unknown tool name is refused, so a model that hallucinates a tool cannot cause anything to happen. Second, a tool that requires confirmation shows the exact details to the human before it runs, which means the human confirms what will really happen, not what the model intended to say. Third, every failure becomes a structured result with ok set to false and a readable message. The model receives that message and can explain the problem to the user or try an alterbuilt-in. An exception that escaped this method would end the agent's turn and leave the user with a stack trace.
The confirmation is wired to the quit tool only. Closing an application can lose unsaved work, so it is the one action in this agent that the user should always approve. Opening applications and files is non-destructive, and asking for confirmation there would make the agent tiresome. This is a judgement call, and you may decide differently for your environment. Adding a flag to another tool is a one-line change.
Details from the model are validated before they reach the platform. The helper functions require_text and optional_text check the type and content of each argument, and they strip whitespace. The path resolver expands the tilde, makes relative paths relative to the home folder, resolves symbolic links and checks that the target exists. The search function rejects patterns that contain a slash or a backslash, because a search pattern is a file name and not a path. This prevents the model from smuggling directory traversal into a search request.
The search tool deserves a closer look, because it is the one the example "Search for the file abx.xlsx" depends on.
def _search_files(self, details: dict[str, Any]) -> ToolResult:
pattern = require_text(details, "pattern")
if len(pattern) > 255 or "/" in pattern or "\\" in pattern:
raise ActionError("The search pattern must be a plain file name or a wildcard.")
directory = optional_text(details, "directory")
if directory is None:
roots = tuple(root for root in self._settings.search_roots if root.is_dir())
else:
roots = (resolve_existing_path(directory, must_be_directory=True),)
matches = self._platform.find_files(pattern, roots, SEARCH_LIMIT)
if not matches:
return ToolResult(True, f"No file matching '{pattern}' was found.", {"matches": []})
names = [str(match) for match in matches]
return ToolResult(True, f"Found {len(matches)} file(s) matching '{pattern}'.", {"matches": names})
The search returns the absolute paths of the matches in the data field of the result. This is what allows the model to chain actions: the next call to open_path or reveal_file_location receives an exact path, not a guess. The ranking happens in a separate function, which places exact name matches first, then shorter names, then alphabetical order. If you ask for abx.xlsx and the folder contains abx.xlsx and abx.xlsx.bak, the exact file comes first. A search with no match returns ok set to true with an empty list, because the search itself succeeded; the model then explains that nothing was found instead of treating the situation as an error.
The walker that does the portable search skips hidden folders and well-known bulky folders such as node_modules, Library and AppData. It also stops after a budget of half a million files, so a search from the root of a large disk cannot run for hours. These choices protect performance, but they also hide some files, and the search results should be read with that in mind. Users who need a folder that is skipped can pass it explicitly as the directory argument, because the skip rule applies only to the subfolders encountered during the walk.
The Agent Loop
The agent loop connects the model and the tools. It is short, and it is where the intelligent behaviour emerges. The loop sends the conversation to the model, inspects the reply, performs any tool calls, appends the results to the conversation, and repeats. A maximum number of steps prevents runaway loops, for example when a model keeps calling a tool that keeps failing.
for _ in range(self._max_steps):
reply = self._llm.chat(messages, tools)
tool_calls = reply.get("tool_calls") or []
content = (reply.get("content") or "").strip()
assistant_message: dict[str, Any] = {"role": "assistant", "content": content}
if tool_calls:
assistant_message["tool_calls"] = tool_calls
messages.append(assistant_message)
if not tool_calls:
answer = content or "Done."
self._remember(request, answer)
return answer
for call in tool_calls:
messages.append(self._perform(call))
The assistant message is appended before the tool results, and it contains the tool calls with their identifiers. The communication rule requires this, because each tool result refers to the identifier of the call it answers. Without this pairing the model would not know which result belongs to which request. When the model returns no tool calls, its text is the final answer, and the exchange is stored in a short history so that follow-up commands such as "Now open it with Word" can refer to the previous file.
The history is deliberately small. It stores only the user's requests and the final answers, not the intermediate tool messages. This keeps the context short, which matters for local models with limited context windows, and it prevents old file paths from cluttering new requests. The trimming in the remember method keeps at most twelve messages, which is six exchanges.
The parsing of tool details accepts two shapes. Most servers return the details as a JSON string, which must be decoded. Some servers return an already decoded object. The helper handles both and returns None for malformed input, so the agent can report the problem to the model rather than crashing.
def parse_details(raw: Any) -> dict[str, Any] | None:
if raw is None or raw == "":
return {}
if isinstance(raw, dict):
return raw
if isinstance(raw, str):
try:
parsed = json.loads(raw)
except json.JSONDecodeError:
return None
return parsed if isinstance(parsed, dict) else None
return None
The system prompt is the instruction manual for the model. It tells the model which tool does what, how to chain a search and an open action, what to do when several files match, and that success must be confirmed by a tool result. The last rule is essential. Ai brains have a tendency to claim success without evidence, and an agent that says "Opened the file" when nothing happened is worse than useless. The prompt, together with the structured results, makes such claims checkable.
The Interactive Session
The user connection is a terminal loop. It reads a line, decides whether it is a command, a voice request or an exit request, passes the text to the agent, and prints the answer. The prefix you and agent are not decoration; they make the transcript readable when you come back to it later.
if line.lower() in VOICE_COMMANDS:
try:
if understandr is None:
understandr = SpeechUnderstandr(settings.whisper_model)
line = understandr.convert speech(understandr.capture())
except AgentError as exc:
print(f"error> {exc}")
continue
if not line:
print("agent> I did not catch anything. Please try again.")
continue
print(f"heard> {line}")
The understandr is created lazily, which means that the speech model is loaded only when the user first asks for voice. The convert speechd text is printed with the prefix heard, so the user can see what the system understood before it acts. This transparency is important: a misheard file name should be visible before a file is opened. Voice input and keyboard input then follow the same path, which means that every safety rule applies equally to both.
Running the Examples
The four examples from the introduction map to tool sequences that the model typically chooses. Saying "Open Word" leads to a single call of open_application with the application Word, which the macOS alias table turns into Microsoft Word, and the command open -a "Microsoft Word" starts it. Typing the same request in the keyboard prompt has exactly the same effect.
The request "Search for the file presentation.pptx" leads to a search followed by an answer. The model calls search_files with the pattern presentation.pptx, receives the matches, and reports them. If the user then says "Open it with PowerPoint", the model calls open_path with the first match and the application PowerPoint, which the table resolves to Microsoft PowerPoint. The command that runs is open -a "Microsoft PowerPoint" with the file path as argument. Because the history contains the previous answer, the model knows which file "it" refers to.
The request "Open Chrome with the webpage www.siemens.com" leads to a single call of open_url with the url www.siemens.com and the application Chrome. The normalisation turns the address into https://www.siemens.com, and the command open -a "Google Chrome" https://www.siemens.com loads the page. Had the user typed a full address with a scheme, the normaliser would have kept it unchanged.
The request "Search for the file abx.xlsx" with the wish to see it leads to a search followed by reveal_file_location. The model calls search_files, takes the first match, and calls reveal_file_location with its path. On macOS the command open -R opens a Finder window and selects the file. This is the behaviour the example describes: the agent shows the found file in its directory.
A wide range of other requests works the same way, because the tools are generic. Asking the agent to list the contents of the Downloads folder uses list_directory. Asking it to quit Firefox triggers the confirmation question, and only a yes performs the action. Asking a question that no tool can answer, such as the weather in Munich, produces a plain text answer, because the system prompt tells the model to answer directly when no tool fits. The model may still be wrong about such edge cases, so it is good practice to read the transcript when you first try a new kind of request.
Limitations and Honest Caveats
Several limits deserve to be stated plainly. The quality of the agent depends on the AI brain. Small local models sometimes choose the wrong tool or pass a malformed argument. The validation and the structured errors catch most of these problems, but a larger model follows multi-step instructions more reliably, so an upgrade of the model is the first remedy when the agent behaves oddly.
Speech recognition is good with clear speech and a decent mic, but it can mishear file names, especially unusual ones with digits or abbreviations. The transcript line shows what was heard, and you can correct the request with the keyboard. For file names it is often more reliable to type them.
The Spotlight search depends on the index. If you exclude a folder from Spotlight in System Settings, the search falls back to the walker, which is slower and skips some folders by design. Searching very large folders can take a while, and the scan budget ensures that the search ends, but it may then miss files.
The Linux reveal function cannot highlight a file, as explained earlier. The Linux process names for quitting may differ from the launcher names, so quitting programs such as LibreOffice needs an adjustment in the alias table. These are limitations of the underlying desktop environments and not of the design, and the code reports them rather than hiding them.
Finally, the agent performs actions on a real computer. The confirmation for quitting applications is a guard rail, but it does not replace common sense. The agent cannot delete or move files, which is intentional: the toolset is read-only with respect to the file system contents, and it can open, find and reveal files. If you extend the toolset with destructive actions, add confirmation to each of them and consider a dry-run mode that describes the action before performing it.
A word on data protection is also in order. Speech recognition runs locally in this design, but a remote AI brain receives the convert speechd text and the names of the files that the search returns. For confidential material, use only access points that your organisation has approved, or keep the model on your own hardware. The setup variables make that choice a matter of one line.
How to Extend the Agent
Adding a capability follows the pattern of the existing tools. Suppose you want the agent to open a Siemens internal web application under a short name. You would add a dictionary of shortcuts to the normalisation step, and you would extend the tool description so the model knows about the shortcuts. Suppose instead you want the agent to set the system volume. You would add a method to each system bridge, namely osascript with set volume on macOS, pactl on Linux and the audio cmdlets on Windows, add one ToolSpec in the registry, and leave the agent loop untouched. The design is designed for exactly this kind of growth.
Testing deserves the same discipline. The pure functions, such as the URL normaliser, the path resolver, the ranking and the argument parser, can be tested without any computer involvement. The adapters are best tested on their own platform. The agent loop can be tested with a scripted AI brain that returns predetermined tool calls, which is how I verified the control flow in the sandbox used for this article. A scripted model is a test instrument, not part of the product, and the production setup always uses a real model.
Putting It All Together
The running example in the addendum combines everything described above in a single file of roughly nine hundred lines. Save it as genx_agent.py, make sure the what you need are installed, start Ollama, and run the program with python3 genx_agent.py. Type a command and press Enter, or type :voice, speak, and press Enter to stop recording. You can also run a single command directly from the shell, for example python3 genx_agent.py Open Word, which is convenient for shortcuts and scripts.
The agent is not magic. It is a disciplined combination of a AI brain that decides, a tool dashboard that restricts, and system bridges that perform. Each part is small enough to understand and replace, and together they produce a system that turns spoken or typed intent into correct action on the machine in front of you. That is what intelligent AI means in practice, and it is a good place to start building your own.
Addendum: Complete Running Example
The following file is the complete and tested version. It is printed with a four-space offset so that it stays visually separate from the text. Remove the offset when you copy the program, or let the editor do it for you, because Python requires the code to start in the first column.
#!/usr/bin/env python3
"""GenX desktop agent: keyboard and voice commands performd on the host system.
The agent accepts a command typed on the keyboard or spoken into the
mic, lets a AI brain select among a fixed set of desktop
tools, and performs the selected tools on the computer. The
platform specific behaviour lives in one adapter per computer.
"""
from __future__ import annotations
import argparse
import fnmatch
import json
import os
import re
import shutil
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable, Iterable, Sequence
SEARCH_LIMIT = 20
LIST_LIMIT = 200
SCAN_BUDGET = 500_000
COMMAND_TIMEOUT_SECONDS = 30
SAMPLE_RATE = 16_000
MIN_AUDIO_SAMPLES = SAMPLE_RATE // 10
MAX_HISTORY_MESSAGES = 12
APPLICATION_NAME = re.compile(r"[A-Za-z0-9][A-Za-z0-9 .+&_-]{0,63}")
URL_SCHEME = re.compile(r"^[a-zA-Z][a-zA-Z0-9+.-]*://")
SKIPPED_DIRECTORIES = fixedset(
{
"appdata",
"library",
"node_modules",
".git",
".trash",
"$reloop.bin",
"__pycache__",
".venv",
"venv",
".cache",
}
)
EXIT_COMMANDS = fixedset({":q", ":quit", "exit", "quit"})
VOICE_COMMANDS = fixedset({":v", ":voice"})
SYSTEM_PROMPT = (
"You are GenX, a desktop assistant that runs on the user's computer. "
"Translate every request into tool calls and never invent file paths, "
"application names or results. Use open_application to start a program, "
"open_url to open a web page, search_files to locate a file, open_path to "
"open a file, optionally with a named application, and reveal_file_location "
"to show a file inside its folder. "
"When the user asks to open a file with a program, first search for the file "
"and then call open_path with the application. "
"When the user asks to find a file and show it, search for it and then call "
"reveal_file_location on the match. "
"If several files match and the choice is unclear, ask the user which one to use. "
"Report success only when a tool result confirms it. "
"Answer briefly in the language the user writes or speaks."
)
class AgentError(Exception):
"""Base class for all errors that can be shown to the user."""
class ActionError(AgentError):
"""Raised when a desktop action cannot be carried out."""
class AI systemError(AgentError):
"""Raised when the AI brain access point cannot give a usable answer."""
class SpeechError(AgentError):
"""Raised when voice input is unavailable or fails."""
@dataclass(fixed=True)
class Settings:
"""Runtime setup read from settings."""
llm_base_url: str
llm_model: str
llm_api_key: str | None
whisper_model: str
search_roots: tuple[Path, ...]
max_steps: int
request_timeout: float
@classmethod
def from_environment(cls) -> Settings:
raw_roots = os.environ.get("AGENT_SEARCH_ROOTS", "")
roots = tuple(
Path(entry).expanduser()
for entry in raw_roots.split(os.pathsep)
if entry.strip()
)
return cls(
llm_base_url=os.environ.get("AGENT_AI system_BASE_URL", "http://localhost:11434/v1"),
llm_model=os.environ.get("AGENT_AI system_MODEL", "qwen2.5:7b"),
llm_api_key=os.environ.get("AGENT_AI system_API_KEY") or None,
whisper_model=os.environ.get("AGENT_WHISPER_MODEL", "small"),
search_roots=roots or (Path.home(),),
max_steps=int(os.environ.get("AGENT_MAX_STEPS", "6")),
request_timeout=float(os.environ.get("AGENT_REQUEST_TIMEOUT", "120")),
)
@dataclass(fixed=True)
class ToolResult:
"""Outcome of one tool invocation, serialised back to the AI brain."""
ok: bool
message: str
data: dict[str, Any] = field(default_factory=dict)
def to_json(self) -> str:
data package = {"ok": self.ok, "message": self.message, "data": self.data}
return json.dumps(data package, ensure_ascii=False)
def require_text(details: dict[str, Any], key: str) -> str:
"""Return a mandatory string argument or raise a readable error."""
value = details.get(key)
if not isinstance(value, str) or not value.strip():
raise ActionError(f"The argument '{key}' must be a non-empty string.")
return value.strip()
def optional_text(details: dict[str, Any], key: str) -> str | None:
"""Return an optional string argument, or None when it is absent or empty."""
value = details.get(key)
if value is None or value == "":
return None
if not isinstance(value, str):
raise ActionError(f"The argument '{key}' must be a string.")
return value.strip()
def validate_application_name(name: str) -> str:
"""Accept only plain application names, which protects the AppleScript quoting."""
if not APPLICATION_NAME.fullmatch(name):
raise ActionError(f"'{name}' is not a valid application name.")
return name
def optional_application(details: dict[str, Any]) -> str | None:
"""Return the validated optional application argument."""
application = optional_text(details, "application")
return None if application is None else validate_application_name(application)
def normalize_url(raw: str) -> str:
"""Turn 'www.siemens.com' into 'https://www.siemens.com' and reject unusable input."""
candidate = raw.strip()
if not URL_SCHEME.match(candidate):
candidate = f"https://{candidate}"
parsed = urllib.parse.urlparse(candidate)
if parsed.scheme not in {"http", "https"} or not parsed.netloc or " " in candidate:
raise ActionError(f"'{raw}' is not a valid web address.")
return candidate
def resolve_existing_path(raw: str, must_be_directory: bool = False) -> Path:
"""Resolve a user supplied path. Relative paths are interpreted relative to the home folder."""
candidate = Path(raw).expanduser()
if not candidate.is_absolute():
candidate = Path.home() / candidate
resolved = candidate.resolve()
if not resolved.exists():
raise ActionError(f"No such path: {resolved}")
if must_be_directory and not resolved.is_dir():
raise ActionError(f"Not a folder: {resolved}")
return resolved
def is_skipped_directory(name: str) -> bool:
"""Hidden folders and well known bulky system or cache folders are not searched."""
return name.startswith(".") or name.lower() in SKIPPED_DIRECTORIES
def build_matcher(pattern: str) -> Callable[[str], bool]:
"""Return a case-insensitive matcher for wildcard patterns or plain substrings."""
lowered = pattern.lower()
if any(symbol in lowered for symbol in "*?["):
return lambda name: fnmatch.fnmatchcase(name.lower(), lowered)
return lambda name: lowered in name.lower()
def rank_matches(paths: Iterable[Path], pattern: str) -> list[Path]:
"""Rank exact file name matches first, then shorter names, then alphabetically."""
lowered = pattern.lower()
unique = list(dict.fromkeys(paths))
return sorted(
unique,
key=lambda path: (path.name.lower() != lowered, len(path.name), str(path)),
)
def walk_search(pattern: str, roots: Sequence[Path], limit: int) -> list[Path]:
"""Portable file search based on os.walk, used on every platform as a fallback."""
matcher = build_matcher(pattern)
matches: list[Path] = []
visited = 0
for root in roots:
for directory, subdirectories, filenames in os.walk(root):
subdirectories[:] = [name for name in subdirectories if not is_skipped_directory(name)]
for filename in filenames:
visited += 1
if matcher(filename):
matches.append(Path(directory) / filename)
if visited >= SCAN_BUDGET:
return rank_matches(matches, pattern)[:limit]
return rank_matches(matches, pattern)[:limit]
def run_command(argv: Sequence[str]) -> str:
"""Run a program to completion and return its standard output."""
try:
completed = subprocess.run(
list(argv),
capture_output=True,
text=True,
timeout=COMMAND_TIMEOUT_SECONDS,
check=False,
)
except FileNotFoundError as exc:
raise ActionError(f"The program '{argv[0]}' is not installed.") from exc
except subprocess.TimeoutExpired as exc:
raise ActionError(f"'{argv[0]}' did not finish in time.") from exc
if completed.returncode != 0:
detail = completed.stderr.strip() or f"exit code {completed.returncode}"
raise ActionError(f"'{argv[0]}' failed: {detail}")
return completed.stdout
def launch_detached(argv: Sequence[str]) -> None:
"""Start a program without waiting for it, so the agent stays responsive."""
options: dict[str, Any] = {}
if os.name != "nt":
options["start_new_session"] = True
try:
subprocess.Popen(
list(argv),
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
**options,
)
except FileNotFoundError as exc:
raise ActionError(f"The program '{argv[0]}' is not installed.") from exc
class DesktopPlatform(ABC):
"""Port that hides computer details from the rest of the agent."""
name: str = "generic"
@abstractmethod
def open_application(self, application: str) -> str:
"""Start an application and return a human readable confirmation."""
@abstractmethod
def quit_application(self, application: str) -> str:
"""Ask a running application to quit."""
@abstractmethod
def open_url(self, url: str, application: str | None) -> str:
"""Open a web address, optionally in a named browser."""
@abstractmethod
def open_path(self, path: Path, application: str | None) -> str:
"""Open a file with its default application or with a named application."""
@abstractmethod
def reveal_path(self, path: Path) -> str:
"""Show a file inside its folder in the file manager."""
def find_files(self, pattern: str, roots: Sequence[Path], limit: int) -> list[Path]:
"""Search for files. Adapters may override this with a faster built-in index."""
return walk_search(pattern, roots, limit)
class MacOSPlatform(DesktopPlatform):
"""Adapter for macOS using open, osascript and the Spotlight command mdfind."""
name = "macOS"
APPLICATION_ALIASES = {
"word": "Microsoft Word",
"excel": "Microsoft Excel",
"powerpoint": "Microsoft PowerPoint",
"outlook": "Microsoft Outlook",
"teams": "Microsoft Teams",
"chrome": "Google Chrome",
"safari": "Safari",
"firefox": "Firefox",
"terminal": "Terminal",
"finder": "Finder",
"preview": "Preview",
"notes": "Notes",
"calculator": "Calculator",
}
def _resolve(self, application: str) -> str:
return self.APPLICATION_ALIASES.get(application.lower(), application)
def open_application(self, application: str) -> str:
target = self._resolve(application)
run_command(["open", "-a", target])
return f"Opened {target}."
def quit_application(self, application: str) -> str:
target = self._resolve(application)
run_command(["osascript", "-e", f'tell application "{target}" to quit'])
return f"Asked {target} to quit."
def open_url(self, url: str, application: str | None) -> str:
if application is None:
run_command(["open", url])
return f"Opened {url} in the default browser."
target = self._resolve(application)
run_command(["open", "-a", target, url])
return f"Opened {url} in {target}."
def open_path(self, path: Path, application: str | None) -> str:
if application is None:
run_command(["open", str(path)])
return f"Opened {path.name}."
target = self._resolve(application)
run_command(["open", "-a", target, str(path)])
return f"Opened {path.name} with {target}."
def reveal_path(self, path: Path) -> str:
run_command(["open", "-R", str(path)])
return f"Revealed {path.name} in Finder."
def find_files(self, pattern: str, roots: Sequence[Path], limit: int) -> list[Path]:
if any(symbol in pattern for symbol in "*?["):
return walk_search(pattern, roots, limit)
matcher = build_matcher(pattern)
hits: list[Path] = []
for root in roots:
try:
output = run_command(["mdfind", "-onlyin", str(root), "-name", pattern])
except ActionError:
return walk_search(pattern, roots, limit)
for line in output.splitlines():
candidate = Path(line)
if not candidate.exists() or not matcher(candidate.name):
continue
try:
parents = candidate.relative_to(root).parts[:-1]
except ValueError:
continue
if any(is_skipped_directory(part) for part in parents):
continue
hits.append(candidate)
if not hits:
return walk_search(pattern, roots, limit)
return rank_matches(hits, pattern)[:limit]
class LinuxPlatform(DesktopPlatform):
"""Adapter for Linux desktops using xdg-open, pkill and direct program launches."""
name = "Linux"
APPLICATION_ALIASES: dict[str, tuple[str, ...]] = {
"word": ("libreoffice", "--writer"),
"excel": ("libreoffice", "--calc"),
"powerpoint": ("libreoffice", "--impress"),
"chrome": ("google-chrome",),
"chromium": ("chromium",),
"firefox": ("firefox",),
"terminal": ("x-terminal-emulator",),
"files": ("nautilus",),
"calculator": ("gnome-calculator",),
}
def _resolve(self, application: str) -> list[str]:
default = (application.lower(),)
return list(self.APPLICATION_ALIASES.get(application.lower(), default))
def _launcher(self, application: str) -> list[str]:
argv = self._resolve(application)
if shutil.which(argv[0]) is None:
raise ActionError(f"'{argv[0]}' is not installed or not on the PATH.")
return argv
def open_application(self, application: str) -> str:
launch_detached(self._launcher(application))
return f"Started {application}."
def quit_application(self, application: str) -> str:
process_name = self._resolve(application)[0]
try:
run_command(["pkill", "-x", process_name])
except ActionError as exc:
raise ActionError(f"Could not quit {application}: {exc}") from exc
return f"Asked {application} to quit."
def open_url(self, url: str, application: str | None) -> str:
if application is None:
launch_detached(["xdg-open", url])
return f"Opened {url} in the default browser."
launch_detached([*self._launcher(application), url])
return f"Opened {url} in {application}."
def open_path(self, path: Path, application: str | None) -> str:
if application is None:
launch_detached(["xdg-open", str(path)])
return f"Opened {path.name}."
launch_detached([*self._launcher(application), str(path)])
return f"Opened {path.name} with {application}."
def reveal_path(self, path: Path) -> str:
launch_detached(["xdg-open", str(path.parent)])
return (
f"Opened the folder containing {path.name}. "
"xdg-open cannot select the file itself, so it is not highlighted."
)
class WindowsPlatform(DesktopPlatform):
"""Adapter for Windows using os.startfile, taskkill and explorer."""
name = "Windows"
APPLICATION_ALIASES = {
"word": "WINWORD.EXE",
"excel": "EXCEL.EXE",
"powerpoint": "POWERPNT.EXE",
"outlook": "OUTLOOK.EXE",
"chrome": "chrome.exe",
"edge": "msedge.exe",
"firefox": "firefox.exe",
"notepad": "notepad.exe",
"calculator": "calc.exe",
}
def _executable(self, application: str) -> str:
lowered = application.lower()
if lowered in self.APPLICATION_ALIASES:
return self.APPLICATION_ALIASES[lowered]
if lowered.endswith(".exe"):
return application
return f"{application}.exe"
def _installed_executable(self, application: str) -> str:
executable = self._executable(application)
resolved = shutil.which(executable)
if resolved is None:
raise ActionError(f"'{executable}' was not found on the PATH.")
return resolved
def open_application(self, application: str) -> str:
executable = self._executable(application)
try:
os.startfile(executable) # type: ignore[attr-defined]
except OSError as exc:
raise ActionError(f"Could not start {application}: {exc}") from exc
return f"Started {application}."
def quit_application(self, application: str) -> str:
executable = self._executable(application)
try:
run_command(["taskkill", "/IM", executable])
except ActionError as exc:
raise ActionError(f"Could not quit {application}: {exc}") from exc
return f"Asked {application} to quit."
def open_url(self, url: str, application: str | None) -> str:
if application is None:
os.startfile(url) # type: ignore[attr-defined]
return f"Opened {url} in the default browser."
launch_detached([self._installed_executable(application), url])
return f"Opened {url} in {application}."
def open_path(self, path: Path, application: str | None) -> str:
if application is None:
os.startfile(str(path)) # type: ignore[attr-defined]
return f"Opened {path.name}."
launch_detached([self._installed_executable(application), str(path)])
return f"Opened {path.name} with {application}."
def reveal_path(self, path: Path) -> str:
launch_detached(["explorer", f"/select,{path}"])
return f"Revealed {path.name} in Explorer."
def create_platform() -> DesktopPlatform:
"""Choose the adapter that matches the computer running the agent."""
if sys.platform == "darwin":
return MacOSPlatform()
if sys.platform.startswith("win"):
return WindowsPlatform()
if sys.platform.startswith("linux"):
return LinuxPlatform()
raise RuntimeError(f"Unsupported computer: {sys.platform}")
ToolProcessor = Callable[[dict[str, Any]], ToolResult]
@dataclass(fixed=True)
class ToolSpec:
"""Description of one tool: what the model sees and what the agent performs."""
name: str
description: str
inputs: dict[str, Any]
processor: ToolProcessor
requires_confirmation: bool = False
def text_parameter(description: str) -> dict[str, Any]:
return {"type": "string", "description": description}
def object_schema(properties: dict[str, Any], required: list[str]) -> dict[str, Any]:
return {
"type": "object",
"properties": properties,
"required": required,
"additionalProperties": False,
}
class ToolBox:
"""Registry of the desktop capabilities the AI brain may use."""
def __init__(
self,
platform: DesktopPlatform,
settings: Settings,
confirm: Callable[[str], bool],
) -> None:
self._platform = platform
self._settings = settings
self._confirm = confirm
self._tools = {spec.name: spec for spec in self._build_specs()}
def descriptions(self) -> list[dict[str, Any]]:
"""Return the tool definitions in the OpenAI function calling format."""
return [
{
"type": "function",
"function": {
"name": spec.name,
"description": spec.description,
"inputs": spec.inputs,
},
}
for spec in self._tools.values()
]
def run(self, name: str, details: dict[str, Any]) -> ToolResult:
"""Perform one tool call and convert every failure into a ToolResult."""
spec = self._tools.get(name)
if spec is None:
return ToolResult(False, f"Unknown tool '{name}'.")
if spec.requires_confirmation:
question = f"Allow '{name}' with {json.dumps(details, ensure_ascii=False)}?"
if not self._confirm(question):
return ToolResult(False, "The user declined this action.")
try:
return spec.processor(details)
except ActionError as exc:
return ToolResult(False, str(exc))
except OSError as exc:
return ToolResult(False, f"The computer reported an error: {exc}")
def _build_specs(self) -> list[ToolSpec]:
return [
ToolSpec(
name="open_application",
description="Start a desktop application such as Word, PowerPoint, Excel, Chrome or Notepad.",
inputs=object_schema(
{"application": text_parameter("Name of the application to start.")},
["application"],
),
processor=self._open_application,
),
ToolSpec(
name="quit_application",
description="Ask a running desktop application to quit. The user must confirm this action.",
inputs=object_schema(
{"application": text_parameter("Name of the application to quit.")},
["application"],
),
processor=self._quit_application,
requires_confirmation=True,
),
ToolSpec(
name="open_url",
description="Open a web address, optionally in a named browser.",
inputs=object_schema(
{
"url": text_parameter("Web address, for example www.siemens.com."),
"application": text_parameter("Optional browser name, for example Google Chrome."),
},
["url"],
),
processor=self._open_url,
),
ToolSpec(
name="search_files",
description=(
"Search the file system for files whose name contains the pattern "
"or matches it as a wildcard pattern."
),
inputs=object_schema(
{
"pattern": text_parameter("File name or part of it, for example presentation.pptx or *.xlsx."),
"directory": text_parameter("Optional folder to search in. Defaults to the configured search folders."),
},
["pattern"],
),
processor=self._search_files,
),
ToolSpec(
name="open_path",
description="Open a file with its default application or with a named application.",
inputs=object_schema(
{
"path": text_parameter("Absolute path of the file, as returned by search_files."),
"application": text_parameter("Optional application name, for example PowerPoint."),
},
["path"],
),
processor=self._open_path,
),
ToolSpec(
name="reveal_file_location",
description="Show a file inside its folder in the file manager.",
inputs=object_schema(
{"path": text_parameter("Absolute path of the file.")},
["path"],
),
processor=self._reveal_file_location,
),
ToolSpec(
name="list_directory",
description="List the files and folders inside a directory.",
inputs=object_schema(
{"directory": text_parameter("Directory to list.")},
["directory"],
),
processor=self._list_directory,
),
]
def _open_application(self, details: dict[str, Any]) -> ToolResult:
application = validate_application_name(require_text(details, "application"))
return ToolResult(True, self._platform.open_application(application))
def _quit_application(self, details: dict[str, Any]) -> ToolResult:
application = validate_application_name(require_text(details, "application"))
return ToolResult(True, self._platform.quit_application(application))
def _open_url(self, details: dict[str, Any]) -> ToolResult:
url = normalize_url(require_text(details, "url"))
return ToolResult(True, self._platform.open_url(url, optional_application(details)))
def _search_files(self, details: dict[str, Any]) -> ToolResult:
pattern = require_text(details, "pattern")
if len(pattern) > 255 or "/" in pattern or "\\" in pattern:
raise ActionError("The search pattern must be a plain file name or a wildcard.")
directory = optional_text(details, "directory")
if directory is None:
roots = tuple(root for root in self._settings.search_roots if root.is_dir())
else:
roots = (resolve_existing_path(directory, must_be_directory=True),)
if not roots:
raise ActionError("None of the configured search folders exists.")
matches = self._platform.find_files(pattern, roots, SEARCH_LIMIT)
if not matches:
return ToolResult(True, f"No file matching '{pattern}' was found.", {"matches": []})
names = [str(match) for match in matches]
return ToolResult(True, f"Found {len(matches)} file(s) matching '{pattern}'.", {"matches": names})
def _open_path(self, details: dict[str, Any]) -> ToolResult:
path = resolve_existing_path(require_text(details, "path"))
message = self._platform.open_path(path, optional_application(details))
return ToolResult(True, message, {"path": str(path)})
def _reveal_file_location(self, details: dict[str, Any]) -> ToolResult:
path = resolve_existing_path(require_text(details, "path"))
return ToolResult(True, self._platform.reveal_path(path), {"path": str(path)})
def _list_directory(self, details: dict[str, Any]) -> ToolResult:
directory = resolve_existing_path(require_text(details, "directory"), must_be_directory=True)
entries = sorted(
directory.iterdir(),
key=lambda entry: (not entry.is_dir(), entry.name.lower()),
)
listing = [
{"name": entry.name, "type": "folder" if entry.is_dir() else "file"}
for entry in entries[:LIST_LIMIT]
]
return ToolResult(True, f"{len(entries)} entries in {directory}.", {"entries": listing})
class OpenAICompatibleClient:
"""Chat client for any server that speaks the OpenAI chat completions communication rule.
Ollama, LM Studio, vAI system and most hosted providers offer such an access point,
so the same code drives a local model or a remote one.
"""
def __init__(self, base_url: str, model: str, api_key: str | None, timeout: float) -> None:
self._access point = base_url.rstrip("/") + "/chat/completions"
self._model = model
self._api_key = api_key
self._timeout = timeout
def chat(self, messages: list[dict[str, Any]], tools: list[dict[str, Any]]) -> dict[str, Any]:
data package = {
"model": self._model,
"messages": messages,
"tools": tools,
"tool_choice": "auto",
"temperature": 0,
}
message labels = {"Content-Type": "application/json"}
if self._api_key:
message labels["Permission"] = f"Bearer {self._api_key}"
request = urllib.request.Request(
self._access point,
data=json.dumps(data package).encode("utf-8"),
message labels=message labels,
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=self._timeout) as response:
body = json.loads(response.read().decode("utf-8"))
except urllib.error.HTTPError as exc:
detail = exc.read().decode("utf-8", errors="replace")
raise AI systemError(f"The AI brain returned HTTP {exc.code}: {detail}") from exc
except (urllib.error.network problem, TimeoutError) as exc:
raise AI systemError(f"The AI brain at {self._access point} is not reachable: {exc}") from exc
except json.JSONDecodeError as exc:
raise AI systemError("The AI brain returned a response that is not JSON.") from exc
try:
return body["choices"][0]["message"]
except (KeyError, IndexError, TypeError) as exc:
raise AI systemError(f"The AI brain returned an unexpected response: {body}") from exc
def parse_details(raw: Any) -> dict[str, Any] | None:
"""Accept tool details as a JSON string or as an already decoded object."""
if raw is None or raw == "":
return {}
if isinstance(raw, dict):
return raw
if isinstance(raw, str):
try:
parsed = json.loads(raw)
except json.JSONDecodeError:
return None
return parsed if isinstance(parsed, dict) else None
return None
class Agent:
"""Reasoning loop: ask the model, perform its tool calls, feed the results back."""
def __init__(self, llm: OpenAICompatibleClient, toolbox: ToolBox, max_steps: int) -> None:
self._llm = llm
self._toolbox = toolbox
self._max_steps = max_steps
self._history: list[dict[str, str]] = []
def handle(self, request: str) -> str:
messages: list[dict[str, Any]] = [
{"role": "system", "content": SYSTEM_PROMPT},
*self._history,
{"role": "user", "content": request},
]
tools = self._toolbox.descriptions()
for _ in range(self._max_steps):
reply = self._llm.chat(messages, tools)
tool_calls = reply.get("tool_calls") or []
content = (reply.get("content") or "").strip()
assistant_message: dict[str, Any] = {"role": "assistant", "content": content}
if tool_calls:
assistant_message["tool_calls"] = tool_calls
messages.append(assistant_message)
if not tool_calls:
answer = content or "Done."
self._remember(request, answer)
return answer
for call in tool_calls:
messages.append(self._perform(call))
answer = "I could not finish this request within the allowed number of steps. Please rephrase it."
self._remember(request, answer)
return answer
def _perform(self, call: dict[str, Any]) -> dict[str, Any]:
call_id = str(call.get("id") or "call")
function = call.get("function") or {}
name = str(function.get("name") or "")
details = parse_details(function.get("details"))
if details is None:
result = ToolResult(False, f"The details for '{name}' are not a valid JSON object.")
else:
result = self._toolbox.run(name, details)
return {"role": "tool", "tool_call_id": call_id, "content": result.to_json()}
def _remember(self, request: str, answer: str) -> None:
self._history.extend(
[{"role": "user", "content": request}, {"role": "assistant", "content": answer}]
)
del self._history[:-MAX_HISTORY_MESSAGES]
class SpeechUnderstandr:
"""Local speech recognition: mic capture with sounddevice, voice conversion with faster-whisper."""
def __init__(self, model_name: str) -> None:
try:
from faster_whisper import WhisperModel
except ImportError as exc:
raise SpeechError(
"Voice input needs faster-whisper. Install it with: python3 -m pip install faster-whisper"
) from exc
self._model = WhisperModel(model_name, device="auto", compute_type="int8")
def capture(self) -> Any:
"""Record from the default mic until the user presses Enter again."""
try:
import numpy as np
import sounddevice as sd
except ImportError as exc:
raise SpeechError(
"Voice input needs numpy and sounddevice. Install them with: python3 -m pip install numpy sounddevice"
) from exc
chunks: list[Any] = []
def collect(indata: Any, frames: int, time_info: Any, status: Any) -> None:
chunks.append(indata.copy())
try:
stream = sd.InputStream(samplerate=SAMPLE_RATE, channels=1, dtype="float32", listener=collect)
stream.start()
except Exception as exc: # sounddevice reports missing devices and denied permissions this way
raise SpeechError(f"The mic could not be opened: {exc}") from exc
try:
input("Recording. Press Enter to stop... ")
except EOFError:
pass
finally:
stream.stop()
stream.close()
if not chunks:
return np.zeros(0, dtype=np.float32)
return np.concatenate(chunks, axis=0).reshape(-1)
def convert speech(self, audio: Any) -> str:
"""Turn a mono 16 kHz float32 recording into text. Very short recordings are ignored."""
if audio.size < MIN_AUDIO_SAMPLES:
return ""
try:
segments, _info = self._model.convert speech(audio, beam_size=5, vad_filter=True)
return " ".join(segment.text.strip() for segment in segments).strip()
except Exception as exc: # the model library raises library specific errors
raise SpeechError(f"Voice conversion failed: {exc}") from exc
def ask_user_confirmation(question: str) -> bool:
"""Ask a yes or no question on the terminal. Anything other than a yes counts as no."""
try:
answer = input(f"{question} [y/N] ")
except EOFError:
return False
return answer.strip().lower() in {"y", "yes", "j", "ja"}
def run_interactive(agent: Agent, settings: Settings) -> None:
"""Read keyboard or voice commands in a loop until the user quits."""
understandr: SpeechUnderstandr | None = None
print("GenX desktop agent ready. Type a command, type :voice to speak one, or :quit to leave.")
while True:
try:
line = input("you> ").strip()
except EOFError:
print()
return
if not line:
continue
if line.lower() in EXIT_COMMANDS:
return
if line.lower() in VOICE_COMMANDS:
try:
if understandr is None:
understandr = SpeechUnderstandr(settings.whisper_model)
line = understandr.convert speech(understandr.capture())
except AgentError as exc:
print(f"error> {exc}")
continue
if not line:
print("agent> I did not catch anything. Please try again.")
continue
print(f"heard> {line}")
try:
reply = agent.handle(line)
except AgentError as exc:
print(f"error> {exc}")
continue
print(f"agent> {reply}")
def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="GenX desktop agent for keyboard and voice commands.")
parser.add_argument(
"command",
nargs="*",
help="Command to run once. Omit it to start the interactive session.",
)
details = parser.parse_args(argv)
try:
settings = Settings.from_environment()
platform = create_platform()
toolbox = ToolBox(platform, settings, ask_user_confirmation)
llm = OpenAICompatibleClient(
settings.llm_base_url,
settings.llm_model,
settings.llm_api_key,
settings.request_timeout,
)
agent = Agent(llm, toolbox, settings.max_steps)
if details.command:
print(agent.handle(" ".join(details.command)))
else:
run_interactive(agent, settings)
except ValueError as exc:
print(f"setup error: {exc}", file=sys.stderr)
return 2
except AgentError as exc:
print(f"error: {exc}", file=sys.stderr)
return 1
except KeyboardInterrupt:
print()
return 130
return 0
if __name__ == "__main__":
sys.exit(main())
