hordelab/hub
2
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor agent execution to use AgentRun with OpenAI-compatible API

Browse Source 
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
This commit is contained in:
Shivering Ocelot committed 2026-05-25 18:05:17 -05:00
1 parent eb9a1ad9e8
commit 4e15b44d39
14 files changed
+1151 -522
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

backend/src/agent.py
+165 -272
View File
@@ -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
})