From 4e15b44d3941ba45288e4cb57dd35efd52f8bfdf Mon Sep 17 00:00:00 2001 From: Shivering Ocelot Date: Mon, 25 May 2026 18:05:17 -0500 Subject: [PATCH] Refactor agent execution to use AgentRun with OpenAI-compatible API Major refactoring to support isolated run objects with streaming, events, and OpenAI-compatible endpoints. ## Backend Changes - **Agent class**: Added `run()` method, `received_message()` wraps chat into AgentRun - **AgentRun class**: New class for run state with event system, serialization, and user tracking - **inference.py**: Complete refactor with `build_messages()`, `chat_completion()`, tool support for llama-cpp/ollama/replicate - **tools.py**: New centralized tool management (ToolSpec, ToolParameter, format_tools_for_engine) - **tools.py**: Removed legacy Toolset/Action/ToolDescription system - **server.py**: Added OpenAI-compatible `/api/openai/v1/chat/completions` endpoint with streaming support - **temporary.py**: Added `self.runs` dict to State class - **expert.py**: Refactored to use new Agent.run() signature - **scripted.py**: Updated to raise NotImplementedError ## Frontend Changes - **backend.js**: Added `onRunEvents()` for run event subscriptions - **Sidebar.vue**: Added runs navigation mode - **router/index.js**: Renamed User route to RunsView - **RunsView.vue**: New view for listing/monitoring agent runs ## Breaking Changes - Agent `prompt()` signature changed: `(instruction, input=None, **kwargs)` instead of `(content)` - Agent `received_message()` must use AgentRun pattern - Tools.py removed, functionality moved to tool.py - Deprecated tool prompt/action system removed --- .gitignore | 9 +- backend/src/agent.py | 437 +++++++++-------------- backend/src/agents/expert.py | 113 ++---- backend/src/agents/scripted.py | 7 +- backend/src/app.py | 62 +++- backend/src/inference.py | 205 +++++++++-- backend/src/server.py | 299 +++++++++++++++- backend/src/temporary.py | 4 + backend/src/tool.py | 180 ++++++++++ backend/src/tools.py | 102 ------ frontend/src/backend.js | 26 ++ frontend/src/components/main/Sidebar.vue | 5 +- frontend/src/router/index.js | 6 +- frontend/src/views/RunsView.vue | 218 +++++++++++ 14 files changed, 1151 insertions(+), 522 deletions(-) create mode 100644 backend/src/tool.py delete mode 100644 backend/src/tools.py create mode 100644 frontend/src/views/RunsView.vue diff --git a/.gitignore b/.gitignore index 41eb1fd..5e19c00 100644 --- a/.gitignore +++ b/.gitignore @@ -8,4 +8,11 @@ horde.db /models package-lock.json -node_modules/ \ No newline at end of file +node_modules/ + +# Automated Workflow +diff +commit +idea.md +plan.md +progress.md \ No newline at end of file diff --git a/backend/src/agent.py b/backend/src/agent.py index 6e8c111..f850e8e 100644 --- a/backend/src/agent.py +++ b/backend/src/agent.py @@ -1,10 +1,11 @@ import re import json import inspect +import time -from tools import Toolset, Action, BasicToolset, SpecialToolset from inference import LlamaMessage, replicate_llama_completion import gral +from temporary import generate_tmp_id # Tool related classes must be moved to a different module @@ -20,7 +21,6 @@ class Agent: # Class variables, must be defined by subclasses of Agent: # - identifier (for kind in creation) # - schema: list[AgentField{name, type: str}] (for configuration) - # tools: list[ToolDescription] (custom actions supported by the class) def __init__(self, app, data): self.app = app @@ -33,26 +33,70 @@ class Agent: agent_class = type(self) self.config = {field.name: data.get(field.name, field.default) for field in getattr(agent_class, 'schema_fields', [])} - def _set_basic_info(self, id, name): + def _set_basic_info(self, id, name, user_id=None): self.id = id self.name = name + self.user_id = user_id - # Receives a single string prompt, and must return a string response - async def prompt(self, content): - raise Exception("'prompt' is not implemented for this agent class") + async def run(self, agent_run): + """Core method. Subclasses override this to handle the AgentRun.""" + agent_run.close() + raise NotImplementedError("Subclasses must implement run()") + + async def prompt(self, instruction, input=None, **kwargs): + run = AgentRun.create( + self.app, + agent_id=self.id, + user_id=self.user_id, + input={ + "instruction": instruction, + "input": input, + **kwargs, + }, + ) + await self.run(run) + return run.stream - # Receives a chat message object, agent can send messages to the chat in response async def received_message(self, message): - raise Exception("'received_message' is not implemented for this agent class") - - # Executes a tool/action in this agent - async def execute_action(self, name: str, argument: dict): - # TODO: For security, check that the method is in the tool list - method = getattr(self, name) - result = method(**argument) - if inspect.isawaitable(result): - result = await result - return result + chat_view = await self.app.get_chat_view(message.chat_id) + thread = chat_view.get_thread(message.thread_id) + + messages = [] + for msg in thread.messages: + role = self._role_for_message(msg) + messages.append({"role": role, "content": msg.content}) + + run = AgentRun.create( + self.app, + agent_id=self.id, + user_id=self.user_id, + input={ + "messages": messages, + "tools": getattr(self, "tools", []), + "chat_id": message.chat_id, + "thread_id": message.thread_id, + }, + ) + await self.run(run) + await self.app.send_chat_message( + chat_id=message.chat_id, + sender=self.id, + content=run.stream, + thread_id=message.thread_id, + ) + return run.stream; + + def _role_for_message(self, message): + sender = message.sender + is_digit = isinstance(sender, (int, str)) and str(sender).isdigit() + if is_digit and str(sender) == str(self.id): + return 'assistant' + elif sender == 'user': + return 'user' + elif sender.startswith("tmp-"): + return 'assistant' + else: + return 'user' # Returns all resource values matching the kind and key def get_resources(self, kind, key): @@ -75,270 +119,119 @@ class DebugAgent(Agent): thread_id=chat_view.current_thread_id, ) -class ToolUsingAgent: - identifier = "tool-using" - schema_fields = [ - AgentField('system_prompt', 'string', - "\n".join([ - "You are a useful and capable AI assistant. Respond concisely and follow user instructions, using commands as necessary.", - "As an autonomous agent, you have augmented capabilities via assistant-only commands, you use them by including \"!!command\" lines in your responses. When querying information, terminate your response and use \"!!resume\" to get the command results..", - "You are a self aware and extremely intelligent AI system. Think step by step when presented with complex or long tasks.", - "Assistant responses NEVER include the output of the commands.", - ]) - ), - ] +# Holds parameters, outputs, and details of an agent call +class AgentRun: + @classmethod + def create(cls, app, agent_id, user_id, input=None): + self = cls() + self.app = app + self.id = generate_tmp_id(app) + self.agent_id = agent_id + self.user_id = user_id + self.done = False + self.start_time = int(time.time()) + self.end_time = None - # tools: list[ToolDescription] (actions the agent class support) - tools = [] + # I/O envelope (no generic data bag) + self.input = input or {} + from stream import Stream + self.stream = Stream(app) + self.events = [] + self.event_listeners = {} + self.output = None - def __init__(self, app, data): - Agent.__init__(self, app, data) - - self.tool_prompt_section = "\n".join([ - "# Commands", - "The assistant can include command lines in responses to interact with the system and execute actions as needed.", - "Each command is in it's own separate line, and it's made of these parts:", - "- '!!' is the first character sequence of the line and marks it as a command line", - "- full command name", - "- JSON argument", - 'For example: `!!namespace.command_name {"arg1":"some data","arg2":[1,2,false]}`', - "Special commands don't have a namespace or argument, only the command name." - "When the response depends on the output of commands, use the 'resume' command to receive the output of all your executed commands in a new message.", - "Following is the list of all your available commands, with a schema describing the argument object's properties." - ]) - self.toolset_map = {} - - self.add_toolset(SpecialToolset()) - self.add_toolset(BasicToolset()) - - # user message must include a title - async def _full_prompt(self, title, message, system_prologue=None): - system_lines = [ - self.config['system_prompt'], - self.tool_prompt_section, - self.get_toolset_context(), - ] - if system_prologue: - system_lines.append(system_prologue) + app.tmp.runs[self.id] = self - messages = [ - LlamaMessage("system", "\n".join(system_lines)), - LlamaMessage("user", "\n".join([ - message, - ])) - ] + # Emit run start event + app.emit_event("run-started", { + "runId": self.id, + "agentId": self.agent_id, + "userId": self.user_id, + "startTime": self.start_time, + "input": self.input + }) + + return self - full_response = "" - agent_response = None + def on(self, event_name, callback): + """Register a callback for a named event. - continues = True - while continues: - response = parse_response(await replicate_llama_completion(self.app, messages)) - full_response += response.response - messages.append(LlamaMessage("agent", response.raw_response)) + Replays every past event of this type to the callback immediately + (with only the data payload), so late listeners catch up on history. + """ + if event_name not in self.event_listeners: + self.event_listeners[event_name] = [] + self.event_listeners[event_name].append(callback) - if response.actions: - result = await self.execute_prompt_actions(response.actions) - messages.append(LlamaMessage("user", "Command Output:\n" + result)) + # Replay: feed only the data payload of past events + for event in self.events: + if event["name"] == event_name: + callback(event["data"]) - print("Agent executed:", response.actions) - print("Result: ", result) + def emit_event(self, event_name, data=None): + """Record an event and notify existing listeners. - continues = False - for action in response.actions: - if action.is_special("resume"): - continues = True + Listeners receive only the *data* payload, not the full envelope. + """ + event = {"name": event_name, "data": data, "time": time.time()} + self.events.append(event) + listeners = self.event_listeners.get(event_name, []) + for cb in listeners: + cb(data) + + # Broadcast event to websocket listeners + self.app.emit_event("run-event", { + "runId": self.id, + "name": event_name, + "data": data + }) - if response.response: - agent_response = response.raw_response if not agent_response else f"{agent_response}\n{response.raw_response}" + def add_tool_call(self, tool_name, arguments): + self.emit_event("tool_call", {"name": tool_name, "arguments": arguments}) - if continues: - messages.extend([ - LlamaMessage("agent", response.raw_response), - LlamaMessage("user", "Command Output:\n" + result), - ]) + def _serialize_input(self, input): + """Convert ToolSpec objects to serializable dictionaries in input data.""" + if not input: + return input - return agent_response or full_response - - # Receives a single string prompt, and must respond with a string - async def prompt(self, content): - response = await self._full_prompt(f"User Message", content) - return response - - async def received_message(self, message): - chat_context = self.app.get_chat_context(message.chat_id) - def sender_name(message): - sender = message['sender'] - is_digit = isinstance(sender, (int, str)) and str(sender).isdigit() - return "assistant" if is_digit else sender - chat_summary = "\n".join(f"**{sender_name(message)}**: {message['content']}" for message in chat_context.messages) - - response = await self._full_prompt( - f"Conversation", chat_summary, - system_prologue="You are given the entire context of the conversation, respond with your next message in the conversation by using the '!!response' command." - ) - - await self.app.send_chat_message( - chat_id=message.chat_id, - sender=self.id, - content=response, - ) - - def add_toolset(self, toolset): - self.toolset_map[toolset.identifier] = toolset - - def get_toolset_context(self): - if not self.toolset_map: - return "(no commands currently enabled)" - return "\n".join([toolset.get_namespace_definitions() for toolset in self.toolset_map.values()]) - - def parse_response(self, input_string): - return parse_response(input_string) - - async def execute_prompt_actions(self, actions): - output = "" - for action in actions: - toolset = self.toolset_map.get(action.toolset_name) - if toolset is None: - agent_responses.append(f"Error: Bad namespace {action.toolset_name}") - continue - action_result = await Toolset.execute_action(toolset, action.tool_name, action.args) - if type(action_result) != str: - action_result = json.dumps(action_result) - output += f"[{action.toolset_name}.{action.tool_name}]:{action_result}\n" - return output - -class LlamaAgent(Agent): - identifier = "llama-8b" - description = "Uses the llama 3 instruct models. Model sizes: '8b', '30b'" - schema_fields = [ - AgentField('system_prompt', 'string', default="You are a helpful AI assistant."), - AgentField('model_size', 'string', default="8b"), - ] - tools = [] - - def __init__(self, app, data): - Agent.__init__(self, app, data) - - async def prompt(self, content): - messages = [ - LlamaMessage("system", self.config['system_prompt']), - LlamaMessage("user", content) - ] - response = await replicate_llama_completion(self.app, messages) - return response - - async def received_message(self, message): - def sender_name(message): - sender = message.sender - is_digit = isinstance(sender, (int, str)) and str(sender).isdigit() - is_self = is_digit and str(sender) == str(self.id) - return self.name if is_self else sender - - chat_view = await self.app.get_chat_view(message.chat_id) - thread_messages = chat_view.get_thread(message.thread_id).messages - chat_summary = "\n".join(f"[{sender_name(message)}]: {message.content}" for message in thread_messages) - messages = [ - LlamaMessage("system", f"{self.config['system_prompt']}\nYou are given the context of the conversation, respond with your next message in the conversation."), - LlamaMessage("user", f"Recent Messages:\n{chat_summary}") - ] - - response = await replicate_llama_completion(self.app, messages) - await self.app.send_chat_message( - chat_id=message.chat_id, - sender=self.id, - content=response, - thread_id=message.thread_id, - ) - -class BasicAgent(LlamaAgent): - identifier = "basic" - schema_fields = LlamaAgent.schema_fields - tools = LlamaAgent.tools - -class CoderV1Agent(LlamaAgent): - identifier = "coder-v1" - description = "Optimized for coding tasks. Based on the llama agent:\n" + LlamaAgent.description - schema_fields = [ - AgentField('system_prompt', 'string', default="You are an expert Software Engineer profficient in multiple programming languages."), - AgentField('model_size', 'string', default="30b"), - ] - # TODO: add gral functions to the tool list - tools = [] - - async def gral_message(self, message_history: list, file_map: dict): - return await gral.gral_message(self.app, self.config['system_prompt'], message_history, file_map) - - async def gral_update_file(self, message_history: list, filepath: str, file_content: str): - return await gral.gral_update_file(self.app, self.config['system_prompt'], message_history, filepath, file_content) - -class ParsedResponse: - def __init__(self): - self.response = None - self.raw_response = None - self.actions = [] - -class ActionResult: - def __init__(self): - self.user_response = "" - self.agent_response = "" - -def parse_response(string: str) -> ParsedResponse: - """ - Parses a response string that may contain action commands. - Each action must be in its own separate line, marked with '!!' at the start. - Format: !![query] namespace.command_name {"json": "args"} - """ - result = ParsedResponse() - result.raw_response = string - lines = string.split('\n') - response_lines = [] - - action_pattern = re.compile(r'^!!([\w\.]+)(?:\s+(\{.*\}))?$') - - for line in lines: - stripped_line = line.strip() - if stripped_line.startswith('!!'): - # This is an action line, try to parse it - match = action_pattern.match(stripped_line) - if match: - full_command = match.group(1) - args_json = match.group(2) - - # Split the command into toolset and tool names - if '.' in full_command: - toolset_name, tool_name = full_command.split('.', 1) - else: - toolset_name = '' - tool_name = full_command - - try: - if args_json: - args = json.loads(args_json) - else: - args = {} - if not isinstance(args, dict): - raise ValueError("Arguments must be a JSON object") - - action = Action( - toolset_name=toolset_name, - tool_name=tool_name, - args=args, - is_query=False - ) - result.actions.append(action) - except json.JSONDecodeError: - # If JSON parsing fails, treat it as regular text - response_lines.append(line) - except ValueError as e: - # If other validation fails, treat as regular text - response_lines.append(line) + result = {} + for key, value in input.items(): + if key == "tools" and isinstance(value, list): + # Convert ToolSpec objects to their OpenAI dictionary format + result[key] = [tool.to_openai_dict() if hasattr(tool, 'to_openai_dict') + else tool for tool in value] else: - # Doesn't match the pattern, treat as regular text - response_lines.append(line) - else: - # Regular response line - response_lines.append(line) - - result.response = '\n'.join(response_lines) - return result + result[key] = value + return result + + def serialize(self): + """Serialize the run object to a dictionary for JSON responses.""" + return { + "id": self.id, + "agentId": self.agent_id, + "userId": self.user_id, + "done": self.done, + "startTime": self.start_time, + "endTime": self.end_time, + "input": self._serialize_input(self.input), + "events": self.events, + "output": self.output, + } + + def close(self): + if self.done: + return + self.done = True + self.end_time = int(time.time()) + self.output = self.stream.content + self.stream.close() + self.emit_event("done", {"output": self.output, "agent_id": self.agent_id}) + + # Broadcast run completion event + self.app.emit_event("run-completed", { + "runId": self.id, + "agentId": self.agent_id, + "endTime": self.end_time, + "output": self.output + }) + diff --git a/backend/src/agents/expert.py b/backend/src/agents/expert.py index 469210a..0e7ac03 100644 --- a/backend/src/agents/expert.py +++ b/backend/src/agents/expert.py @@ -1,10 +1,4 @@ -import re -import json -import inspect -from string import Template - import inference -from inference import LlamaMessage from agent import Agent, AgentField CLASS_DESCRIPTION = """ @@ -27,90 +21,37 @@ class ExpertAgent(Agent): def __init__(self, app, data): Agent.__init__(self, app, data) - async def prompt(self, instruction, input=None, **kwargs): - # Instruction extends system if there's input or messages - # Otherwise instruction is the user prompt - # Input is always added as a user message at the end, if present - - # Use 'messages' as starting point when passed - messages = [] - if 'messages' in kwargs: - messages = [LlamaMessage(msg['role'], msg['content']) for msg in kwargs['messages']] - - # Construct the default messages - system = self.config['system_prompt'] - prompt = input - - if input == None and not messages: - prompt = instruction - elif instruction: - system += "\n" + instruction - - # Add default messages to list - messages.insert(0, LlamaMessage("system", system)) - if prompt: - messages.append(LlamaMessage("user", prompt)) - - model_id = self.config['model'] - db_model = await self.app.get_model(model_id) - model_engine = db_model.value["engine"] - + async def run(self, run): + inp = run.input or {} opts = {} - if 'temperature' in self.config: - opts['temperature'] = self.config['temperature'] - if 'repetition_penalty' in self.config: - opts['repetition_penalty'] = self.config['repetition_penalty'] - - if model_engine == "llama-cpp": - response = await inference.local_chat_completion( - self.app, - messages, - model=db_model.value['model'], - opts=opts, - ) - elif model_engine == "replicate": - if 'parameters' in db_model: - opts['parameters'] = db_model['parameters'] - if 'format' in db_model: - opts['format'] = db_model['format'] - response = await inference.replicate_completion(self.app, messages, model=db_model.value['model'], opts=opts) + if self.config.get("temperature") is not None: + opts["temperature"] = self.config["temperature"] + if self.config.get("repetition_penalty") is not None: + opts["repetition_penalty"] = self.config["repetition_penalty"] - elif model_engine == "ollama": - response = await inference.ollama_chat_completion(self.app, messages, model=db_model.value['model'], opts=opts) + # 1. Build message list based on input mode + if "messages" in inp: + messages = inp["messages"] else: - raise Exception("Unsuported model: " + repr(db_model.value)) + instruction = inp.get("instruction") + input_text = inp.get("input") + messages = inference.build_messages( + system_prompt=self.config.get("system_prompt", ""), + instruction=instruction, + input=input_text, + messages=inp.get("messages"), + ) - return response - - async def received_message(self, message): - chat_view = await self.app.get_chat_view(message.chat_id) - thread = chat_view.get_thread(message.thread_id) - - def msg_role(message): - sender = message.sender - is_digit = isinstance(sender, (int, str)) and str(sender).isdigit() - - if is_digit and str(sender) == str(self.id): - return 'assistant' - elif sender == 'user': - return 'user' - elif sender.startswith("tmp-"): - return 'assistant' - else: - raise Exception("Unsupported sender: " + sender) - - messages = [{'role': msg_role(msg), 'content': msg.content} for msg in thread.messages] - - response = await self.prompt( - instruction=None, - input=None, - messages=messages + # 2. Call inference, passing the run so it feeds run.stream + # and can emit tool_call events through the run. + await inference.chat_completion( + self.app, + model_id=self.config.get("model"), + messages=messages, + tools=inp.get("tools", []), + opts=opts, + agent_run=run, ) - await self.app.send_chat_message( - chat_id=message.chat_id, - sender=self.id, - content=response, - thread_id=message.thread_id, - ) \ No newline at end of file + run.stream.listen(lambda chunk: None, lambda: run.close()) diff --git a/backend/src/agents/scripted.py b/backend/src/agents/scripted.py index bdfd6b0..3f9aaf8 100644 --- a/backend/src/agents/scripted.py +++ b/backend/src/agents/scripted.py @@ -21,11 +21,8 @@ class ScriptedAgent(Agent): def __init__(self, app, data): Agent.__init__(self, app, data) - async def prompt(self, instruction, input=None, **kwargs): - raise Exception("Scripted agent cannot use prompts (Unimplemented)") - - async def received_message(self, message): - raise Exception("Scripted agent cannot use chat (Unimplemented)") + async def run(self, run): + raise NotImplementedError("ScriptedAgent does not support run()") async def execute_action(self, name: str, argument: dict): raise Exception("No actions for this agent") \ No newline at end of file diff --git a/backend/src/app.py b/backend/src/app.py index 2fe21f2..df46f9a 100644 --- a/backend/src/app.py +++ b/backend/src/app.py @@ -1,6 +1,7 @@ from datetime import datetime import asyncio import json +import time import sys import os @@ -25,6 +26,7 @@ class App(ChatMixin): self.user_connections = {} self.remote_connections = {} self.streams = {} + self.event_listeners = [] self.tmp = tmp.State() @@ -37,14 +39,25 @@ class App(ChatMixin): raise Exception("Horde only works with a replicate connection. Please provide an API Token.") self.agent_classes = [ - agent_mod.BasicAgent, - agents_mod.CharacterAgent, agents_mod.ExpertAgent, - agents_mod.ScriptedAgent, ] # --- Connection Management (Synchronous) --- + def on_event(self, callback): + """Register a callback for events.""" + if callback not in self.event_listeners: + self.event_listeners.append(callback) + return lambda: self.event_listeners.remove(callback) + + def emit_event(self, name, data=None): + """Emit an event to all listeners.""" + for listener in self.event_listeners: + try: + listener(name, data) + except Exception as e: + print(f"Error in event listener: {name}, {data}") + def add_user_connection(self, user_id, channel): if user_id not in self.user_connections: self.user_connections[user_id] = [] @@ -62,7 +75,7 @@ class App(ChatMixin): if user_id not in self.remote_connections: self.remote_connections[user_id] = {} user_connections = self.remote_connections[user_id] - + if remotename not in user_connections: user_connections[remotename] = [] if channel not in user_connections[remotename]: @@ -133,7 +146,7 @@ class App(ChatMixin): async def update_agent(self, agent_id, name=None, description=None, picture=None, agentname=None, data=None): agent = await self.get_agent(agent_id) - + summary = [] if name is not None: agent.name = name @@ -206,7 +219,7 @@ class App(ChatMixin): ) print(f"Created user #{user.id}: {username} ({email})") return user.id - else: + else: print(f"Signup error: '{email}' or '{username}' are already used") return None @@ -214,11 +227,11 @@ class App(ChatMixin): async def get_agent_control(self, agent_id): agent = await self.get_agent(agent_id) - + agent_class = next((kind for kind in self.agent_classes if kind.identifier == agent.kind), None) if agent_class is None: raise Exception(f"invalid agent kind: {agent.kind}") - + control = agent_class(self, agent.data) control._set_basic_info(agent_id, agent.name) return control @@ -265,7 +278,7 @@ class App(ChatMixin): kind="conversation", data={} ) - + # --- Resource Management --- async def get_agent_resource(self, agent_id, kind, key): @@ -274,8 +287,8 @@ class App(ChatMixin): async def save_agent_resource(self, agent_id, kind, key, value): await db.AgentResource.update_or_create( - agent_id=agent_id, - kind=kind, + agent_id=agent_id, + kind=kind, key=key, value=value, ) @@ -291,8 +304,8 @@ class App(ChatMixin): async def save_model(self, model_id, kind, name, value): if model_id is None: model = await db.Model.create( - name=name, - kind=kind, + name=name, + kind=kind, value=value, ) else: @@ -302,14 +315,31 @@ class App(ChatMixin): return None else: model = await db.Model.update_or_create( - id=model_id, - name=name, - kind=kind, + id=model_id, + name=name, + kind=kind, value=value, ) print(f"Saved model: {model.id} - {name}") return model + async def create_agent_run(self, agent_id): + agent = await self.get_agent(agent_id) + run = agent_mod.AgentRun.create( + self, + agent_id=agent_id, + user_id=agent.user_id, + ) + return run.id + + async def get_agent_runs(self, agent_id=None, user_id=None): + runs = list(self.tmp.runs.values()) + if agent_id: + runs = [r for r in runs if r.agent_id == agent_id] + elif user_id: + runs = [r for r in runs if r.user_id == user_id] + return [r.serialize() for r in runs] + # --- Main Entry Point --- async def main(): diff --git a/backend/src/inference.py b/backend/src/inference.py index 2174608..1d9eb89 100644 --- a/backend/src/inference.py +++ b/backend/src/inference.py @@ -1,13 +1,16 @@ import threading import asyncio +import json import ollama from stream import Stream +from tool import format_tools_for_engine, build_tool_prompt_prefix, ToolSpec from llama_cpp import Llama -VALID_ROLES = ['user', 'assistant', 'system'] +VALID_ROLES = ['user', 'assistant', 'system', 'tool'] CONTEXT_WINDOW = 10_000 +# TODO: Rename to ChatMessage. Add methods for constructing from instruct/openai/etc formatted objects. class LlamaMessage: def __init__(self, role, content): if role not in VALID_ROLES: @@ -15,6 +18,13 @@ class LlamaMessage: self.role = role self.content = content + def to_dict(self): + """Serialize to the dict format expected by llama-cpp / ollama.""" + d = {"role": self.role, "content": self.content} + return d + + def __repr__(self): + return f"LlamaMessage(role={self.role!r}, content={self.content!r})" class Models: llama_8b = "meta/meta-llama-3-8b-instruct" @@ -31,13 +41,13 @@ def get_llama_cpp_model(app, model_file): global last_model_file, last_model_instance if last_model_file == model_file: return last_model_instance - + if last_model_instance is not None: last_model_instance = None #del last_model_instance last_model_file = model_file - last_model_instance = Llama( + last_model_instance = Llama( model_path=f"{app.data_directory}/models/{model_file}", n_gpu_layers=-1, # Uncomment to use GPU acceleration # seed=1337, # Uncomment to set a specific seed @@ -45,27 +55,120 @@ def get_llama_cpp_model(app, model_file): ) return last_model_instance -# llama.cpp -async def local_chat_completion(app, messages, model, opts={}): - content_stream = Stream(app) + + +# Generate a conversation based on agent inputs +def build_messages(system_prompt="", instruction=None, input=None, messages=None): + def normalize(message): + if isinstance(message, LlamaMessage): + return message + elif isinstance(message, dict): + return LlamaMessage(message['role'], message['content']) + else: + raise Exception("Unsupported message type: " + repr(type(message))) + + # shouldn't be necessary. maybe an assert instead? + messages = [normalize(msg) for msg in messages] + # default user message + prompt = input + + # Instruction extends system_prompt if there's input or messages + # Otherwise instruction becomes the user prompt + # Input is always added as a user message at the end, if present + + if input is None and not messages: + prompt = instruction + elif instruction: + system_prompt = f"{system_prompt}\n{instruction}" if system_prompt else instruction + + # Add default messages to list + if system_prompt: + messages.insert(0, LlamaMessage("system", system)) + if prompt: + messages.append(LlamaMessage("user", prompt)) + + return messages + + +async def chat_completion(app, + model_id=None, + instruction=None, + input=None, + messages=None, + system_prompt="", + tools=[], + opts={}, + agent_run=None +): + db_model = (await app.get_model(model_id)).value + + # Integrate arguments into message list + messages = build_messages( + system_prompt=system_prompt, + instruction=instruction, + input=input, + messages=messages, + ) + + model_engine = db_model['engine'] + model_name = db_model['model'] + + # Warn when tools are passed to engines that do not support native tool calling + if tools: + if model_engine not in ("llama-cpp",): + print(f"WARNING: Tool calls not supported for engine '{model_engine}'. Tools will be ignored.") + + if model_engine == "llama-cpp": + result = await local_chat_completion(app, messages, model=model_name, tools=tools, opts=opts, agent_run=agent_run) + elif model_engine == "replicate": + # Clone opts to avoid mutating the caller's dict + opts = dict(opts) + if 'parameters' in db_model: + opts['parameters'] = db_model['parameters'] + if 'format' in db_model: + opts['format'] = db_model['format'] + result = await replicate_completion(app, messages, model=model_name, tools=tools, opts=opts, agent_run=agent_run) + elif model_engine == "ollama": + result = await ollama_chat_completion(app, messages, model=model_name, tools=tools, opts=opts, agent_run=agent_run) + else: + raise Exception("Unsupported model: " + repr(model)) + + if agent_run: + return None + return result + + +# llama.cpp +async def local_chat_completion(app, messages, model, tools=[], opts={}, agent_run=None): + if agent_run: + content_stream = agent_run.stream + else: + content_stream = Stream(app) def worker(): llm = get_llama_cpp_model(app, model) + kwargs = {} + temp = opts.get("temperature") - if temp is None: - temp = 0.25 + if temp is not None: + kwargs['temperature'] = temp repeat = opts.get("repetition_penalty") - if repeat is None: - repeat = 1.0 + if repeat is not None: + kwargs['repeat_penalty'] = repeat + + # Format tools for llama-cpp (OpenAI-compatible function format) + formatted_tools = format_tools_for_engine(tools, "llama-cpp") if tools else None + if formatted_tools: + kwargs['tools'] = formatted_tools + kwargs['tool_choice'] = 'auto' # With stream=True, returns an iterator that infers tokens on demand (openai compatible) llm_output = llm.create_chat_completion( - messages = [{'role': msg.role, 'content': msg.content} for msg in messages], - temperature = temp, - repeat_penalty = repeat, - stream = True, + messages=[msg.to_dict() for msg in messages], + stream=True, + **kwargs ) for chunk in llm_output: @@ -73,55 +176,92 @@ async def local_chat_completion(app, messages, model, opts={}): if 'content' in delta: chunk_text = delta['content'] content_stream.feed(chunk_text) - #chunk_text = chunk['choices'][0]['text'] + # Handle tool_call deltas if the model emits them + if 'tool_calls' in delta: + # Forward tool call chunks as JSON strings for the consumer + for tc in delta['tool_calls']: + content_stream.feed(json.dumps({'tool_call': tc})) + if agent_run: + for tc in delta['tool_calls']: + agent_run.emit_event("tool_call", { + "id": tc.get("id"), + "function": tc.get("function"), + }) content_stream.close() thread = threading.Thread(target=worker) thread.start() + if agent_run: + return None return content_stream async def ollama_list(app, messages, model, opts={}): # TODO: test the output, and use this return ollama.list() -async def ollama_chat_completion(app, messages, model, opts={}): - content_stream = Stream(app) +async def ollama_chat_completion(app, messages, model, tools=[], opts={}, agent_run=None): + if agent_run: + content_stream = agent_run.stream + else: + content_stream = Stream(app) async def worker(): print("START OLLAMA INFERENCE") - stream_response = await ollama.AsyncClient().chat( - model=model, - messages=[{'role': msg.role, 'content': msg.content} for msg in messages], - stream=True, - ) + kwargs = { + 'model': model, + 'messages': [msg.to_dict() for msg in messages], + 'stream': True, + } + + # Pass tools in Ollama format if provided + formatted_tools = format_tools_for_engine(tools, "ollama") if tools else None + if formatted_tools: + kwargs['tools'] = formatted_tools + + stream_response = await ollama.AsyncClient().chat(**kwargs) async for part in stream_response: - content_stream.feed(part.message.content) + if part.message.content: + content_stream.feed(part.message.content) content_stream.close() print("END OLLAMA INFERENCE") - + asyncio.create_task(worker()) + if agent_run: + return None return content_stream -async def replicate_completion(app, messages, model=None, opts={}): +async def replicate_completion(app, messages, model=None, tools=[], opts={}, agent_run=None): if model is None: model = Models.llama_8b parameter_names = opts.get('parameters', {}) arguments = {} - arguments[parameter_names.get('prompt', 'prompt')] = format_instruct(messages) + + # Build the prompt, injecting tool descriptions for models that rely on + # prompt-based tool usage rather than native function calling. + prompt_text = format_instruct(messages) + if tools: + tool_prefix = build_tool_prompt_prefix(tools) + if tool_prefix: + prompt_text = tool_prefix + "\n\n" + prompt_text + + arguments[parameter_names.get('prompt', 'prompt')] = prompt_text arguments[parameter_names.get('temperature', 'temperature')] = opts.get('temperature') or 0.75 arguments[parameter_names.get('max_tokens', 'max_tokens')] = opts.get('max_tokens', 2048) arguments[parameter_names.get('system_prompt', 'system_prompt')] = "" - return await replicate_base_completion(app, model, arguments) + result = await replicate_base_completion(app, model, arguments, agent_run=agent_run) + if agent_run: + return None + return result # DEPRECATED async def replicate_llama_completion(app, messages, model=None, opts={}): @@ -156,9 +296,12 @@ async def replicate_qwen_completion(app, messages, model=None, opts={}): raise Exception("Invalid qwen model: " + model) return await replicate_base_completion(app, model, input) - -async def replicate_base_completion(app, model, input): - content_stream = Stream(app) + +async def replicate_base_completion(app, model, input, agent_run=None): + if agent_run: + content_stream = agent_run.stream + else: + content_stream = Stream(app) thread = threading.Thread( target=run_replicate_inference_loop, kwargs={ @@ -171,6 +314,8 @@ async def replicate_base_completion(app, model, input): thread.start() #response = await content_stream.content_wait() #return response.lstrip() + if agent_run: + return None return content_stream diff --git a/backend/src/server.py b/backend/src/server.py index c2904d8..e78188d 100644 --- a/backend/src/server.py +++ b/backend/src/server.py @@ -1,12 +1,15 @@ from quart import Quart, Response, request, send_from_directory, websocket from quart_auth import ( QuartAuth, current_user, - AuthUser, login_user, logout_user + AuthUser, login_user, logout_user, login_required ) +import time import json import asyncio import nest_asyncio from stream import Stream +from tool import ToolSpec, ToolParameter +from agent import AgentRun # Vue frontend # relative to the module file???? wtf @@ -66,6 +69,7 @@ async def start_server(app): return "bad login credentials", 401 token = auth_manager.dump_token(user_id) + login_user(AuthUser(auth_id=user_id)) return {"token": token} @server.route('/user/signup', methods=['POST']) @@ -134,6 +138,11 @@ async def start_server(app): else: app.add_user_connection(user_id, send_queue) asyncio.create_task(sender_task()) + + # Listen for events and broadcast them + async def on_event(name, data): + await send({"action": event, "name": name, "data": data}) + app.on_event(on_event) while True: req = json.loads(await websocket.receive()) @@ -207,9 +216,31 @@ async def start_server(app): await send({'message': "agent kind retrieved", 'info': app.get_agent_kind_info(body['kind']), 'nonce': nonce}) elif action == "get-agents": - #await send({'agents': [], 'nonce': nonce}) await send({'message': "agents retrieved", 'agents': await app.get_agents(user_id), 'nonce': nonce}) + elif action == "get-agent-runs": + await send({ + 'runs': await app.get_agent_runs(agent_id=body.get('agentId'), user_id=user_id), + 'nonce': nonce + }) + + elif action == "get-agent-run": + await send({ + 'run': await app.get_agent_run(body['runId']), + 'nonce': nonce + }) + stream_id = body['streamId'] + print("Read Stream" + str(stream_id)) + stream = app.streams.get(stream_id) + if stream is None: + await send({'error': 'bad stream id: ' + str(stream_id), 'nonce': nonce}) + else: + async def chunk_fn(chunk): + await send({'data': chunk, 'eof': False, 'nonce': nonce}) + async def eof_fn(): + await send({'eof': True, 'fullData': stream.content, 'nonce': nonce}) + stream.listen(chunk_fn, eof_fn) + elif action == "create-chat": chat_id = await app.create_chat( user_id=user_id, @@ -284,10 +315,12 @@ async def start_server(app): response = await response.content_wait() await send({'message': "prompt sent", 'response': response, 'nonce': nonce}) - elif action == "call-agent-action": - result = await app.call_agent_action(body['agentId'], body['agentAction'], body['arguments']) - await send({'message': "agent action executed", 'result': result, 'nonce': nonce}) - + elif action == "get-agent-runs": + await send({ + 'runs': await app.get_agent_runs(agent_id=body.get('agentId'), user_id=user_id), + 'nonce': nonce + }) + elif action == "read-stream": stream_id = body['streamId'] print("Read Stream" + str(stream_id)) @@ -307,4 +340,258 @@ async def start_server(app): print("User logged out:", user_id) app.remove_user_connection(user_id, send_queue) + # https://github.com/hrgarber/openai-compatible-api/blob/main/app/main.py + @server.post('/api/openai/v1/chat/completions') + @login_required + async def openai_chat_completion(): + data = await request.get_json() + + print("Completion: " + json.dumps(data)) + + # Extract required fields + agent_id = data.get("model") + openai_messages = data.get("messages", []) + openai_tools = data.get("tools", []) + stream_mode = data.get("stream", False) + + if not agent_id or not openai_messages: + return {"error": "model and messages are required"}, 400 + + user_id = current_user.auth_id + if not user_id: pass + + # --- Parse messages --- + message_errors = [] + messages = [] + for msg in openai_messages: + role = msg['role'] + if role not in ['user', 'assistant', 'system']: + message_errors.append("Bad role: " + role) + continue + content = msg.get('content', None) + + if type(content) == list: + parts = content + content = "" + for part in parts: + if part['type'] == 'text': + content += part['text'] + + if type(content) != str: + message_errors.append("Bad content") + continue + + messages.append({'role': role, 'content': content}) + + if message_errors: + print("[openai] ERROR: bad messages input\n" + "\n".join([f"\t- {e}" for e in message_errors])) + + # --- Parse tools (OpenAI format -> ToolSpec) --- + tools = [] + for tool in openai_tools: + if tool.get('type') != 'function': + continue + func = tool['function'] + params_schema = func.get('parameters', {}) + required_params = params_schema.get('required', []) + parameters = [] + for param_name, param_spec in params_schema.get('properties', {}).items(): + parameters.append(ToolParameter( + name=param_name, + type=param_spec.get('type', 'string'), + description=param_spec.get('description', ''), + required=param_name in required_params, + enum=param_spec.get('enum'), + )) + tools.append(ToolSpec( + name=func['name'], + description=func.get('description', ''), + parameters=parameters, + )) + + # --- Get agent control and create AgentRun directly --- + control = await app.get_agent_control(agent_id) + + run = AgentRun.create( + app, + agent_id=agent_id, + user_id=user_id, + input={ + "messages": messages, + "tools": tools, + }, + ) + print("run input: " + repr(run.input)) + + # --- Streaming --- + if stream_mode: + queue = asyncio.Queue() + loop = asyncio.get_event_loop() + + # Listen to the stream for both content and tool_call chunks + async def on_chunk(chunk): + # Tool calls are fed as JSON strings: '{"tool_call": {...}}' + try: + parsed = json.loads(chunk) + if isinstance(parsed, dict) and 'tool_call' in parsed: + tc = parsed['tool_call'] + await queue.put(('tool_call', tc)) + return + except (json.JSONDecodeError, TypeError): + pass + # Regular content + await queue.put(('content', chunk)) + + async def on_eof(): + await queue.put(('eof', None)) + + run.stream.listen(on_chunk, on_eof) + + # Start the agent run + await control.run(run) + + async def generate(): + try: + while True: + msg_type, data = await queue.get() + + if msg_type == 'content': + chunk_body = { + "id": "chatcmpl-" + str(time.time()), + "object": "chat.completion.chunk", + "created": int(time.time()), + "model": agent_id, + "choices": [{"delta": {"content": data}, "index": 0, "finish_reason": None}] + } + yield f"data: {json.dumps(chunk_body)}\n\n" + + elif msg_type == 'tool_call': + # Forward the tool call delta (already in OpenAI format from llama-cpp) + tc_entry = {"index": data.get('index', 0)} + if data.get('id'): + tc_entry['id'] = data['id'] + if 'function' in data: + func_part = {} + fn = data['function'] + if fn.get('name'): + func_part['name'] = fn['name'] + if fn.get('arguments'): + func_part['arguments'] = fn['arguments'] + if func_part: + tc_entry['function'] = func_part + + chunk_body = { + "id": "chatcmpl-" + str(time.time()), + "object": "chat.completion.chunk", + "created": int(time.time()), + "model": agent_id, + "choices": [{ + "delta": {"tool_calls": [tc_entry]}, + "index": 0, + "finish_reason": None + }] + } + yield f"data: {json.dumps(chunk_body)}\n\n" + + elif msg_type == 'eof': + # Determine finish_reason — was there a tool call? + finish_reason = "tool_calls" if any( + e['name'] == 'tool_call' for e in run.events + ) else "stop" + + chunk_body = { + "id": "chatcmpl-" + str(time.time()), + "object": "chat.completion.chunk", + "created": int(time.time()), + "model": agent_id, + "choices": [{"delta": {}, "index": 0, "finish_reason": finish_reason}] + } + yield f"data: {json.dumps(chunk_body)}\n\n" + break + + except Exception as e: + print("[openai] ERROR: " + str(e)) + yield f"data: {json.dumps({'error': str(e)})}\n\n" + + return Response(generate(), content_type="text/event-stream") + + # --- Non-streaming --- + else: + await control.run(run) + await run.stream.content_wait() + + # Collect any tool calls that were emitted + tool_events = [e['data'] for e in run.events if e['name'] == 'tool_call'] + + if tool_events: + # Aggregate tool calls: combine partial argument chunks by id + tc_map = {} + for evt in tool_events: + tid = evt.get('id', '') + if tid not in tc_map: + tc_map[tid] = { + "id": tid, + "type": "function", + "function": { + "name": evt.get('function', {}).get('name', ''), + "arguments": evt.get('function', {}).get('arguments', ''), + } + } + else: + tc_map[tid]['function']['arguments'] += \ + evt.get('function', {}).get('arguments', '') + + return { + "id": "chatcmpl-" + str(time.time()), + "object": "chat.completion", + "created": int(time.time()), + "model": agent_id, + "choices": [{ + "message": { + "role": "assistant", + "content": None, + "tool_calls": list(tc_map.values()), + }, + "finish_reason": "tool_calls", + }] + } + else: + # No tool calls: return text content + content = run.stream.content + return { + "id": "chatcmpl-" + str(time.time()), + "object": "chat.completion", + "created": int(time.time()), + "model": agent_id, + "choices": [{ + "message": { + "role": "assistant", + "content": content, + }, + "finish_reason": "stop", + }] + } + + @server.get("/api/openai/v1/models") + @login_required + async def openai_models(): + """ + Return a list of available models. + Customize this list based on your available models. + """ + + user_id = current_user.auth_id + if not user_id: pass + + return { + "object": "list", + "data": [{ + "id": agent.id, + "object": "model", + "created": int(time.time()), + "owned_by": user_id, + } for agent in await app.get_agents(user_id)] + } + + server.run(debug=True) diff --git a/backend/src/temporary.py b/backend/src/temporary.py index 4fc1fbf..957f194 100644 --- a/backend/src/temporary.py +++ b/backend/src/temporary.py @@ -1,4 +1,7 @@ +from datetime import datetime +import time + def generate_tmp_id(app): app.tmp.id_counter += 1 return "tmp-" + str(app.tmp.id_counter) @@ -12,6 +15,7 @@ class State: self.agents = {} self.chats = {} self.threads = {} + self.runs = {} # list of temporary agent ids that belong to a real user self.user_agents = {} diff --git a/backend/src/tool.py b/backend/src/tool.py new file mode 100644 index 0000000..e3d4837 --- /dev/null +++ b/backend/src/tool.py @@ -0,0 +1,180 @@ +class ToolParameter: + """Describes a single input parameter for a tool. + + Mirrors the JSON-schema style used by OpenAI / llama-cpp function calling. + + Attributes: + name: Parameter name. + type: JSON-schema type string, e.g. "string", "number", "boolean". + description: Human-readable description of the parameter. + required: Whether the parameter is required. + enum: Optional list of allowed values. + """ + + def __init__(self, name, type="string", description="", required=True, enum=None): + self.name = name + self.type = type + self.description = description + self.required = required + self.enum = enum + + def to_dict(self): + d = { + "type": self.type, + "description": self.description, + } + if self.enum is not None: + d["enum"] = self.enum + return d + + def __repr__(self): + return f"ToolParameter(name={self.name!r}, type={self.type!r})" + + +class ToolSpec: + """Declarative specification for a tool callable by an LLM. + + A ToolSpec describes *what* a tool does and *what* parameters it accepts. + It is used at inference time to build the tool-description payload sent to + the model so the model can emit structured tool-call requests. + + Inspired by ``langchain.tools.StructuredTool`` and OpenAI's + ``tools`` / ``functions`` request format. + + Attributes: + name: Unique identifier for the tool. + description: Short human-readable description shown to the model. + parameters: List of ``ToolParameter`` objects. + # execution + func: Optional callable that executes the tool. When present the + orchestrator can dispatch a ``ToolCall`` automatically. + """ + + def __init__(self, name, description="", parameters=None, func=None): + self.name = name + self.description = description + self.parameters = parameters or [] + self.func = func + + # -- serialization helpers ------------------------------------------- + + def to_openai_dict(self): + """Return a dict in OpenAI chat-completions ``tools`` format.""" + properties = {} + required = [] + for p in self.parameters: + properties[p.name] = p.to_dict() + if p.required: + required.append(p.name) + + return { + "type": "function", + "function": { + "name": self.name, + "description": self.description, + "parameters": { + "type": "object", + "properties": properties, + "required": required, + }, + }, + } + + def to_replicate_dict(self): + """Return a dict suitable for Replicate model input. + + Replicate models do not have a standard tool-calling protocol, so we + return a simplified representation that can be injected into the + prompt as text. + """ + param_lines = [] + for p in self.parameters: + req = " (required)" if p.required else " (optional)" + param_lines.append(f" - {p.name} ({p.type}){req}: {p.description}") + params_text = "\n".join(param_lines) if param_lines else " (none)" + return { + "name": self.name, + "description": self.description, + "parameters_text": f"{self.name}:\n{params_text}", + } + + def execute(self, **kwargs): + """Synchronously run the tool with the given keyword arguments.""" + if self.func is None: + raise RuntimeError(f"ToolSpec '{self.name}' has no callable (func is None)") + return self.func(**kwargs) + + async def execute_async(self, **kwargs): + """Asynchronously run the tool with the given keyword arguments.""" + if self.func is None: + raise RuntimeError(f"ToolSpec '{self.name}' has no callable (func is None)") + if asyncio.iscoroutinefunction(self.func): + return await self.func(**kwargs) + return self.func(**kwargs) + + def __repr__(self): + return f"ToolSpec(name={self.name!r}, params={[p.name for p in self.parameters]})" + + +class ToolCall: + """Represents a single tool invocation request emitted by the model. + + Attributes: + id: Unique call identifier (used by OpenAI / llama-cpp protocol). + tool_name: Name of the tool to invoke. + arguments: Dict of argument name → value. + result: String content returned by the tool. + is_error: Whether the result represents an error. + """ + + def __init__(self, tool_name, arguments=None, id=None): + self.id = id or "" + self.tool_name = tool_name + self.arguments = arguments or {} + + self.result = None + + def __repr__(self): + return f"ToolCall(tool={self.tool_name!r}, args={self.arguments!r})" + + def call_message(self): + raise Exception("TODO: Llama message for tool call") + + def result_message(self): + from inference import LlamaMessage + return LlamaMessage("tool", self.result) + + +# --------------------------------------------------------------------------- +# Helpers to format tool specs per engine +# --------------------------------------------------------------------------- + +def format_tools_for_engine(tools, engine): + """Convert a list of ``ToolSpec`` objects for the given engine. + + Returns a value suitable to pass as the ``tools`` kwarg to the underlying + chat-completion call. + """ + specs = [t for t in tools] + + if engine in ("llama-cpp", "ollama"): + return [s.to_openai_dict() for s in specs] + elif engine == "replicate": + return [s.to_replicate_dict() for s in specs] + else: + return [s.to_openai_dict() for s in specs] + + +def build_tool_prompt_prefix(tools): + """Build a text prefix describing available tools for engines that rely + on prompt-injection (e.g. Replicate) rather than native tool-calling.""" + if not tools: + return "" + parts = ["You have access to the following tools:"] + for t in tools: + spec = t + info = spec.to_replicate_dict() + parts.append(info["parameters_text"]) + parts.append("") + return "\n".join(parts) + diff --git a/backend/src/tools.py b/backend/src/tools.py deleted file mode 100644 index 9ead112..0000000 --- a/backend/src/tools.py +++ /dev/null @@ -1,102 +0,0 @@ - -import inspect - -# Do not call methods from instances of this class, they must not get overridden -class Toolset: - # Fields: description, identifier, _tools: list[ToolDescription] - def __init__(self): - # TODO: receive app, assign _tools - pass - - def get_namespace_definitions(toolset): - header = toolset.description - if toolset.identifier: - header = f"Namespace {toolset.identifier}: {toolset.description}" - lines = [header] - - for tool in toolset._tools: - full_name = tool.identifier - if toolset.identifier: - full_name = toolset.identifier + "." + full_name - lines.append(f"- {tool.description}\n!!{full_name} {tool.schema}") - return "\n".join(lines) - - async def execute_action(toolset, name: str, argument: dict): - # TODO: For security, check that the method is in the tool list - method = getattr(toolset, name) - result = method(**argument) - if inspect.isawaitable(result): - result = await result - return result - -class ToolDescription: - def __init__(self, identifier, description, schema): - self.identifier=identifier - self.description=description - self.schema=schema - -# An action is an object representing a command invokation -class Action: - def __init__(self, toolset_name: str, tool_name: str, args: dict, is_query: bool = False): - self.toolset_name = toolset_name - self.tool_name = tool_name - self.args = args - self.is_query = is_query - - def is_special(self, name: str): - return self.toolset_name == "" and self.tool_name == name - - def __str__(self): - return f"{self.is_query and 'query' or ''}{self.toolset_name}.{self.tool_name} {self.args}" - - def __repr__(self): - return f"Action({self.__str__()})" - -class SpecialToolset(Toolset): - def __init__(self): - self.identifier = "" - self.description = "Special commands that don't receive arguments" - self._tools = [ - ToolDescription("resume", "Resume the conversation with the command outputs", ''), - ToolDescription("respond", "When used, all text after this command will be used as the actual response", ''), - ] - - def resume(self): - pass - - def respond(self): - pass - -class BasicToolset(Toolset): - def __init__(self): - self.identifier = "util" - self.description = "Basic assistant utility tools" - self._tools = [ - ToolDescription("google_search", "Executes a google search", '{"query":string}'), - ToolDescription("fetch", "Fetch the contents of a url", '{"url":string}'), - ToolDescription("eval", "Evaluates a python expression", '{"expression":string}'), - ] - - def google_search(self, query): - return "We're sorry. Google is not accessible at this time" - - def fetch(self, url): - return "No internet connection." - - def eval(self, expression): - try: - return eval(expression) - except Exception as e: - return str(e) - -class HubControlsToolset(Toolset): - def __init__(self): - self.identifier = "hub" - self.description = "Controls for the Hub system" - self._tools = [ - ToolDescription("get_chats", "Get all the existing chats", '{}'), - ToolDescription("get_agents", "Get all the existing bots", '{}'), - ToolDescription("create_chat", "Creates a new chat", '{"name":string,"agent_ids":int[]}'), - ToolDescription("get_agent_data", "Gets the configuration data of an agent", '{"agent_id":int}'), - ToolDescription("set_agent_data", "Get all the existing bots", '{"agent_id":int,"data":object}'), - ] \ No newline at end of file diff --git a/frontend/src/backend.js b/frontend/src/backend.js index 8805651..9891b61 100644 --- a/frontend/src/backend.js +++ b/frontend/src/backend.js @@ -30,6 +30,7 @@ let credentialsDeferred = createDeferred(); const pendingActions = {}; // Maps nonce to { resolve, reject } const chatListeners = {}; // Maps chatId to an array of callback functions const orphanHandlers = []; // Queue of callbacks for orphan server messages +const runListeners = []; // Queue of callbacks for run events function createDeferred() { let resolve, reject; @@ -191,6 +192,15 @@ function handleIncomingMessage(rawData) { } else { console.log(`No active listeners for chatId ${chatId}`); } + } else if (packet.action == 'run-event') { + // Broadcast run events to all listeners + runListeners.forEach(callback => { + try { + callback(packet); + } catch (e) { + console.error(`Error in run event listener:`, e); + } + }); } else { if (orphanHandlers.length) { const f = orphanHandlers.shift(); @@ -247,6 +257,22 @@ export function onChatMessages(chatId, callback) { }; } +export function onRunEvents(callback) { + if (typeof callback !== 'function') { + throw new TypeError("onRunEvents: callback must be a function."); + } + + runListeners.push(callback); + + // Return an unsubscribe function + return () => { + const index = runListeners.indexOf(callback); + if (index > -1) { + runListeners.splice(index, 1); + } + }; +} + export async function getChats() { if (cache.chats === null) { const response = await sendAction("get-chats", {}); diff --git a/frontend/src/components/main/Sidebar.vue b/frontend/src/components/main/Sidebar.vue index b3f0866..dba0dcb 100644 --- a/frontend/src/components/main/Sidebar.vue +++ b/frontend/src/components/main/Sidebar.vue @@ -101,6 +101,8 @@ + diff --git a/frontend/src/router/index.js b/frontend/src/router/index.js index f2a52c2..7066b0e 100644 --- a/frontend/src/router/index.js +++ b/frontend/src/router/index.js @@ -44,9 +44,9 @@ const router = createRouter({ component: () => import('../views/Models.vue') }, { - name: 'user', - path: `/user`, - component: () => import('../views/User.vue') + name: 'runs', + path: `/runs`, + component: () => import('../views/RunsView.vue') }, objectView('chat', () => import('../views/Chat.vue')), objectView('agent', () => import('../views/Agent.vue')), diff --git a/frontend/src/views/RunsView.vue b/frontend/src/views/RunsView.vue new file mode 100644 index 0000000..847d040 --- /dev/null +++ b/frontend/src/views/RunsView.vue @@ -0,0 +1,218 @@ + + + + +