From d934dc0ef7d6b19f9592371d1af53517ece0fec4 Mon Sep 17 00:00:00 2001 From: Shivering Date: Sun, 21 Jun 2026 05:44:15 -0500 Subject: [PATCH] WIP Stash --- backend/requirements.txt | 4 +- backend/src/agent.py | 39 ++++--- backend/src/agents/expert.py | 62 ++++++++--- backend/src/inference.py | 136 ++++++++++++++++++------- backend/src/migrations/0001_initial.py | 2 +- backend/src/server.py | 2 +- backend/src/tool.py | 4 +- frontend/src/runs.js | 2 - frontend/src/views/RunsView.vue | 2 +- 9 files changed, 175 insertions(+), 78 deletions(-) diff --git a/backend/requirements.txt b/backend/requirements.txt index 68c51f4..98eb2d2 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -5,7 +5,7 @@ sqlalchemy nest_asyncio replicate qdrant-client[fastembed] -ollama +#ollama tortoise-orm>=1.0.0 -llama-cpp-python +#llama-cpp-python openrouter \ No newline at end of file diff --git a/backend/src/agent.py b/backend/src/agent.py index f3d8ef2..19137a7 100644 --- a/backend/src/agent.py +++ b/backend/src/agent.py @@ -2,6 +2,7 @@ import re import inspect import time import asyncio +import traceback from temporary import generate_tmp_id from stream import Stream @@ -66,7 +67,18 @@ class Agent: }, ) await self.run(run) - return run.get_output_stream() + out_stream = Stream(self.app) + + def on_output_event(stream): + def on_chunk(chunk): + out_stream.feed(chunk) + stream.listen(on_chunk, lambda: None) + def on_done(data): + out_stream.close() + run.on('output', on_output_event) + run.on('done', on_done) + + return out_stream async def received_message(self, message): chat_view = await self.app.get_chat_view(message.chat_id) @@ -213,7 +225,8 @@ class AgentRun: cb(data) except Exception as e: print(f"ERROR: handling event '{event_name}' in run {self.id}\n" + str(e)) - + print(traceback.format_exc()) + # Broadcast event to websocket listeners self.app.emit_event("run-event", { "runId": self.id, @@ -221,24 +234,6 @@ class AgentRun: "data": data }) - def get_output_stream(self): - if not self.output: - self.output = Stream(self.app) - self.emit_event('output', self.output) - return self.output - - def feed_output(self, chunk): - stream = self.get_output_stream() - if chunk: - self.output.feed(chunk) - - def feed_reasoning(self, chunk): - if not self.reasoning: - self.reasoning = Stream(self.app) - self.emit_event('reasoning', self.reasoning) - if chunk: - self.reasoning.feed(chunk) - def add_tools(self, tools): if "tools" in self.input: self.input["tools"] += tools @@ -298,7 +293,7 @@ class AgentRun: self.close() # Emit cancellation event - self.emit_event("cancelled", {"runId": self.id, "agentId": self.agent_id}) + self.emit_event("cancelled", {}) # Broadcast cancellation event self.app.emit_event("run-cancelled", { @@ -314,4 +309,4 @@ class AgentRun: self.end_time = int(time.time()) if self.output: self.output.close() if self.reasoning: self.reasoning.close() - self.emit_event("done", {"output": self.output.content, "agent_id": self.agent_id}) + self.emit_event("done", {}) diff --git a/backend/src/agents/expert.py b/backend/src/agents/expert.py index 5b25b8c..a3f5b72 100644 --- a/backend/src/agents/expert.py +++ b/backend/src/agents/expert.py @@ -1,6 +1,8 @@ import inference from agent import Agent, AgentField +import asyncio + CLASS_DESCRIPTION = """ Expert AI with customizable system prompt. Will get guidelines, documents, and other advanced features in the future. @@ -22,16 +24,18 @@ class ExpertAgent(Agent): async def run(self, run): opts = {} - agent_tools = self.tools - print('agent tools ' + repr(agent_tools)) - run.add_tools(agent_tools) + using_tools = False + if 'tools' not in run.input: + if self.tools: + run.input['tools'] = self.tools + using_tools = True 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"] - # 1. Build message list based on input + # Build message list based on input messages = inference.build_messages( system_prompt=self.config.get("system_prompt", ""), instruction=run.input.get("instruction"), @@ -39,14 +43,44 @@ class ExpertAgent(Agent): messages=run.input.get("messages"), ) - # 2. Call inference, passing the run to feed run.stream and tool calls - await inference.chat_completion( - self.app, - model_id=self.config.get("model"), - messages=messages, - tools=run.input.get("tools", []), - opts=opts, - agent_run=run, - ) + # Collect inference output. These are resused across calls + tool_calls = [] + output_stream = None + inference_end = asyncio.Event() + def on_output(stream): + nonlocal output_stream + output_stream = stream + run.on('output', on_output) + run.on('inference-end', lambda event: inference_end.set()) + run.on('tool-call', lambda tc: tool_calls.append(tc)) - run.get_output_stream().listen(lambda chunk: None, lambda: run.close()) + while True: + # Reset data + inference_end.clear() + tool_calls = [] + output_stream = None + + # Call inference, passing the run to get events out + await inference.chat_completion( + self.app, + model_id=self.config.get("model"), + messages=messages, + tools=run.input.get("tools", []), + opts=opts, + agent_run=run, + ) + + await inference_end.wait() + + messages.append(inference.LlamaMessage('assistant', output_stream and await output_stream.content_wait() or "")) + + if using_tools and tool_calls: + for tc in tool_calls: + tool = next(t for t in self.tools if t.name == tc.tool_name) + if not tool: + print(f"Bad agent tool name: {tc.name}") + continue + tc.result = await tool.execute_async(**tc.arguments) + messages.append(inference.LlamaMessage('tool', tc)) + else: + break diff --git a/backend/src/inference.py b/backend/src/inference.py index b430506..bb16f55 100644 --- a/backend/src/inference.py +++ b/backend/src/inference.py @@ -2,12 +2,12 @@ import threading import asyncio import json import os -import ollama +#import ollama import traceback from stream import Stream from tool import format_tools_for_engine, build_tool_prompt_prefix, ToolSpec, ToolCall -from llama_cpp import Llama +#from llama_cpp import Llama from openrouter import OpenRouter VALID_ROLES = ['user', 'assistant', 'system', 'tool'] @@ -35,12 +35,60 @@ class Models: qwen3_8b = "lucataco/qwen3-vl-8b-instruct:39e893666996acf464cff75688ad49ac95ef54e9f1c688fbc677330acc478e11" qwen3_235b = "qwen/qwen3-235b-a22b-instruct-2507" +def format_openai(messages): + formatted = [] + last_assistant_message = None + for msg in messages: + if msg.role == 'tool': + if not last_assistant_message: + raise Exception("tool message before any assistant message") + tool_calls = last_assistant_message.get('tool_calls') + if tool_calls == None: + tool_calls = [] + last_assistant_message['tool_calls'] = tool_calls + tool_calls.append({ + 'id': msg.content.id, + 'type': 'function', + 'function': { + 'name': msg.content.tool_name, + 'arguments': json.dumps(msg.content.arguments), + }, + }) + formatted.append({ + 'role': 'tool', + 'tool_call_id': msg.content.id, + 'content': json.dumps(msg.content.output), + }) + else: + fmsg = {'role': msg.role, 'content': msg.content} + formatted.append(fmsg) + if msg.role == 'assistant': + last_assistant_message = fmsg + return formatted + + # llama cpp models model_cache = {} last_model_file = None last_model_instance = None +def safe_queue(callback): + queue = asyncio.Queue() + loop = asyncio.get_running_loop() + async def read_loop(): + while True: + args = await queue.get() + await callback(*args) + loop.create_task(read_loop()) + + def send(*args): + def queue_put(): + loop.create_task(queue.put(args)) + loop.call_soon_threadsafe(queue_put) + return send + def get_llama_cpp_model(app, model_file): + raise Exception("llama-cpp disabled") global last_model_file, last_model_instance if last_model_file == model_file: return last_model_instance @@ -95,24 +143,13 @@ def build_messages(system_prompt="", instruction=None, input=None, messages=None 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'] @@ -140,10 +177,22 @@ async def chat_completion(app, # llama.cpp async def local_chat_completion(app, messages, model, tools=[], opts={}, agent_run=None): - if agent_run: - content_stream = agent_run.get_output_stream() - else: - content_stream = Stream(app) + if not agent_run: + raise Exception("agent_run required") + + content_stream = None + async def message_loop(name, value=None): + nonlocal content_stream, agent_run + if name == 'output': + if content_stream == None: + content_stream = Stream(app) + agent_run.emit_event('output', content_stream) + content_stream.feed(value) + elif name == 'close': + if content_stream: + content_stream.close() + agent_run.emit_event('inference-end', {}) + send_message = safe_queue(message_loop) def worker(): llm = get_llama_cpp_model(app, model) @@ -179,21 +228,16 @@ async def local_chat_completion(app, messages, model, tools=[], opts={}, agent_r delta = chunk['choices'][0]['delta'] if 'content' in delta: - chunk_text = delta['content'] - content_stream.feed(chunk_text) + send_message('output', delta['content']) # 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"), - }) + send_message("tool-call", { + "id": tc.get("id"), + "function": tc.get("function"), + }) - content_stream.close() + send_message('close') thread = threading.Thread(target=worker) if agent_run: @@ -209,6 +253,7 @@ async def ollama_list(app, messages, model, opts={}): return ollama.list() async def ollama_chat_completion(app, messages, model, tools=[], opts={}, agent_run=None): + raise Exception("ollama disabled") if agent_run: content_stream = agent_run.stream else: @@ -259,6 +304,29 @@ async def openrouter_chat_completion(app, messages, model, tools=[], opts={}, ag if not agent_run: raise Exception("agent_run required") + # TODO: Separate as InferenceRun class + content_stream = None + reasoning_stream = None + async def message_loop(name, value=None): + nonlocal content_stream, reasoning_stream, agent_run + if name == 'output': + if content_stream == None: + content_stream = Stream(app) + agent_run.emit_event('output', content_stream) + content_stream.feed(value) + if name == 'reasoning': + if reasoning_stream == None: + reasoning_stream = Stream(app) + agent_run.emit_event('output', reasoning_stream) + reasoning_stream.feed(value) + if name == 'tool-call': + agent_run.emit_event('tool-call', value) + elif name == 'close': + if content_stream: content_stream.close() + if reasoning_stream: reasoning_stream.close() + agent_run.emit_event('inference-end', {}) + emit = safe_queue(message_loop) + async def worker(): print("START OPENROUTER INFERENCE") if not app.openrouter_client: @@ -266,7 +334,7 @@ async def openrouter_chat_completion(app, messages, model, tools=[], opts={}, ag kwargs = { 'model': model, - 'messages': [msg.to_dict() for msg in messages], + 'messages': format_openai(messages), 'stream': True, } @@ -294,9 +362,9 @@ async def openrouter_chat_completion(app, messages, model, tools=[], opts={}, ag delta = event.choices[0].delta if delta.content: - agent_run.feed_output(delta.content) + emit('output', delta.content) if delta.reasoning: - agent_run.feed_reasoning(delta.reasoning) + emit('reasoning', delta.reasoning) if delta.tool_calls: print("[openrouter] tool call:" + repr(delta.tool_calls)) for tc in delta.tool_calls: @@ -325,9 +393,9 @@ async def openrouter_chat_completion(app, messages, model, tools=[], opts={}, ag try: arguments = json.loads(tc.arguments) tc.arguments = arguments - agent_run.emit_event('tool-call', tc) + emit('tool-call', tc) except Exception: - print("[openrouter] Bad tool call arguments") + print("[openrouter] Bad tool call arguments for " + repr(tc)) print(traceback.format_exc()) except asyncio.CancelledError: print(f"[openrouter] task cancelled for run {agent_run.id}") @@ -336,7 +404,7 @@ async def openrouter_chat_completion(app, messages, model, tools=[], opts={}, ag print(f"[openrouter] exception during inference") print(traceback.format_exc()) - agent_run.close() + emit('close') print("END OPENROUTER INFERENCE") task = asyncio.create_task(worker()) diff --git a/backend/src/migrations/0001_initial.py b/backend/src/migrations/0001_initial.py index 5a96d21..6bb7f4d 100644 --- a/backend/src/migrations/0001_initial.py +++ b/backend/src/migrations/0001_initial.py @@ -1,6 +1,6 @@ from tortoise import migrations from tortoise.migrations import operations as ops -from orjson import loads +from json import loads from tortoise.fields.base import OnDelete from tortoise.fields.data import JSON_DUMPS from tortoise import fields diff --git a/backend/src/server.py b/backend/src/server.py index 14c6fa7..2dc7d18 100644 --- a/backend/src/server.py +++ b/backend/src/server.py @@ -587,4 +587,4 @@ async def start_server(app): } - server.run(debug=True) + await server.run_task(debug=True) diff --git a/backend/src/tool.py b/backend/src/tool.py index 7f052d5..463f031 100644 --- a/backend/src/tool.py +++ b/backend/src/tool.py @@ -1,4 +1,5 @@ import inspect +import asyncio from typing import get_type_hints class ToolParameter: @@ -225,7 +226,8 @@ class ToolCall: self.result = None def __repr__(self): - return f"ToolCall(tool={self.tool_name!r}, args={self.arguments!r})" + id_prefix = self.id + ', ' if self.id else '' + return f"ToolCall({id_prefix}tool={self.tool_name!r}, args={self.arguments!r})" def call_message(self): raise Exception("TODO: Llama message for tool call") diff --git a/frontend/src/runs.js b/frontend/src/runs.js index 67db622..73fcb83 100644 --- a/frontend/src/runs.js +++ b/frontend/src/runs.js @@ -67,8 +67,6 @@ function updateRun(event) { } } else if (name == 'done') { run.done = true; - run.endTime = data.endTime; - run.output = data.output; } else if (name == 'chat-message') { let chatListeners = listeners.chats[data.chatId] || []; for (let f of chatListeners) { diff --git a/frontend/src/views/RunsView.vue b/frontend/src/views/RunsView.vue index c845c5e..b8677f3 100644 --- a/frontend/src/views/RunsView.vue +++ b/frontend/src/views/RunsView.vue @@ -41,7 +41,7 @@ }); function isStreamEvent(event) { - return typeof event.data == "object" && event.data.streamId != null; + return event.data instanceof Object && event.data.streamId != null; } function getStreamRef(event) {