From 0dee6a392ffdd454c3f02210281a987c70bf85d0 Mon Sep 17 00:00:00 2001 From: Shivering Ocelot Date: Sat, 10 Oct 2026 12:38:01 -0500 Subject: [PATCH] feat: send agent prompt -> prompt agent --- backend/src/agent.py | 65 +++++++++++++---------- backend/src/app.py | 3 +- backend/src/app_mixins/agent.py | 11 ++-- backend/src/features/web.py | 61 ++++++++++++++++++++++ backend/src/server/openai.py | 91 +++++++++++++++++---------------- backend/src/server/websocket.py | 40 ++++++--------- frontend/src/views/Agent.vue | 33 +++++++++--- 7 files changed, 199 insertions(+), 105 deletions(-) create mode 100644 backend/src/features/web.py diff --git a/backend/src/agent.py b/backend/src/agent.py index fcab85f..5db56a4 100644 --- a/backend/src/agent.py +++ b/backend/src/agent.py @@ -5,7 +5,6 @@ import asyncio import traceback from temporary import generate_tmp_id -from stream import Stream # Tool related classes must be moved to a different module @@ -66,7 +65,10 @@ class Agent: agent_run.close() raise NotImplementedError("Subclasses must implement run()") - # Deprecate this. This is just run with extra baggage. + # Async generator: constructs a run, executes it, and yields every message + # the run emits (raw inference.Message objects: role + content) in arrival + # order. No projection: no delta collapsing, no tool filtering, no dropping + # of reasoning/usage. Consumers route or accumulate as they see fit. async def prompt(self, instruction, input=None, **kwargs): run = AgentRun.create( self.app, @@ -79,39 +81,46 @@ class Agent: }, ) - done_event = asyncio.Event() - out_stream = Stream(self.app) - tool_calls = [] - tool_names = None - if 'tools' in kwargs: - tool_names = [t['function']['name'] for t in kwargs['tools'] if isinstance(t, dict)] + queue = asyncio.Queue() - def on_chunk(chunk): - out_stream.feed(chunk) def on_inference(msg): - kwargs = {'content': msg.content} + queue.put_nowait(('message', msg)) - if msg.role == 'assistant': - msg.content.listen(on_chunk, lambda: None) - elif msg.role == 'tool' and tool_names: - if msg.content.tool_name in tool_names: - tool_calls.append(msg.content) - def on_done(ev): - done_event.set() - out_stream.close() + def on_done(_data): + queue.put_nowait(('done', None)) + # Register listeners before scheduling the task. AgentRun.on replays + # past events of that name, so this both avoids a race and guarantees + # no early messages are lost. run.on('inference', on_inference) run.on('done', on_done) - await self.run(run) - if tool_names: - await done_event.wait() - return { - 'output': out_stream, - 'tool_calls': tool_calls, - } - else: - return out_stream + run.task = asyncio.create_task(self.run(run)) + + def on_task_done(task): + # Safety net: an agent that raises (or forgets run.close()) would + # otherwise leave the consumer awaiting forever. + queue.put_nowait(('task', task)) + + run.task.add_done_callback(on_task_done) + + try: + while True: + kind, payload = await queue.get() + if kind == 'message': + yield payload + elif kind == 'done': + break + else: # task + if payload.cancelled(): + break + exc = payload.exception() + if exc: + raise exc + break + finally: + if not run.task.done(): + run.task.cancel() async def received_message(self, message): diff --git a/backend/src/app.py b/backend/src/app.py index 313c7b7..998eaff 100644 --- a/backend/src/app.py +++ b/backend/src/app.py @@ -47,7 +47,8 @@ class App(ChatMixin, AgentMixin): ] from features.calc import Calc - self.tool_classes = [Calc] + from features.web import WebSearch, WebFetch + self.tool_classes = [Calc, WebSearch, WebFetch] # --- Connection Management (Synchronous) --- diff --git a/backend/src/app_mixins/agent.py b/backend/src/app_mixins/agent.py index 38df2fa..84bde1d 100644 --- a/backend/src/app_mixins/agent.py +++ b/backend/src/app_mixins/agent.py @@ -180,10 +180,15 @@ class AgentMixin: runs = [r for r in runs if r.user_id == user_id] return [r.serialize() for r in runs] - async def send_agent_prompt(self, agent_id, instruction, input=None, options={}): + async def prompt_agent(self, agent_id, instruction, input=None, options={}): + # Mirror agent.prompt's generator so run ownership and lifecycle stay + # inside the mixin: callers just iterate and receive the same raw + # inference.Message objects. get_agent_control is awaited lazily on first + # iteration, keeping the control object alive for the generator's + # lifetime. control = await self.get_agent_control(agent_id) - response = await control.prompt(instruction, input, **options) - return response + async for msg in control.prompt(instruction, input, **options): + yield msg async def call_agent_action(self, agent_id, action, args): control = await self.get_agent_control(agent_id) diff --git a/backend/src/features/web.py b/backend/src/features/web.py new file mode 100644 index 0000000..0e85ed8 --- /dev/null +++ b/backend/src/features/web.py @@ -0,0 +1,61 @@ +"""Web tools: online search and page fetching. + +Both tools subclass :class:`~backend.src.tool.Tool` and rely on the base +class's automatic ``ToolSpec`` generation (docstring -> description, +``run`` signature -> parameters). +""" + +import re +import urllib.request +from html import unescape + +from tool import Tool + + +def _strip_html(html: str) -> str: + """Light HTML filter: drop scripts/styles, strip tags, collapse whitespace.""" + html = re.sub( + r"<(script|style|noscript)[^>]*>.*?", + " ", + html, + flags=re.DOTALL | re.IGNORECASE, + ) + text = re.sub(r"<[^>]+>", " ", html) + text = unescape(text) + return re.sub(r"\s+", " ", text).strip() + + +class WebSearch(Tool): + """Search the web with DuckDuckGo. Returns titles, URLs and snippets.""" + + def run(self, query: str, max_results: int = 5): + # `ddgs` is the current package name; `duckduckgo_search` is the legacy one. + try: + from ddgs import DDGS + except ImportError: # pragma: no cover - fallback for older installs + from duckduckgo_search import DDGS + + with DDGS() as ddgs: + results = list(ddgs.text(query, max_results=max_results)) + + return [ + { + "title": r.get("title", ""), + "url": r.get("href", ""), + "snippet": r.get("body", ""), + } + for r in results + ] + + +class WebFetch(Tool): + """Fetch a web page and return its text content, lightly filtered.""" + + def run(self, url: str): + req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) + with urllib.request.urlopen(req, timeout=15) as resp: + html = resp.read().decode("utf-8", errors="replace") + return _strip_html(html) + + +__all__ = ["WebSearch", "WebFetch"] \ No newline at end of file diff --git a/backend/src/server/openai.py b/backend/src/server/openai.py index 15fc853..db09708 100644 --- a/backend/src/server/openai.py +++ b/backend/src/server/openai.py @@ -2,7 +2,6 @@ import json, time, traceback, asyncio, uuid from quart import Response, request from quart_auth import login_required, current_user from tool import ToolSpec, ToolParameter, ToolCall -from agent import AgentRun from utils import encode_json # --------------------------------------------------------------------------- @@ -62,43 +61,49 @@ async def execute_deferred(tc): tc.result = await tc.tool.execute_async(**tc.arguments) -def attach_run_handlers(run, tool_names): - """Wire the run's output events into a single queue. +def messages_to_queue(generator, tool_names): + """Drive a prompt generator into a single queue. The queue receives ('content'|'reasoning'|'tool_call'|'deferred_call', data) tuples and terminates with ('eof', None). Tool calls are routed (client vs - native/deferred) here so both response modes share the same logic. + native/deferred) here so both response modes share the same logic. The + returned ``state`` dict tracks whether any tool call was observed, which + drives ``finish_reason``. """ queue = asyncio.Queue() loop = asyncio.get_event_loop() + state = {'tool_call': False, 'error': None} async def on_content(chunk): await queue.put(('content', chunk)) async def on_reasoning(chunk): await queue.put(('reasoning', chunk)) - async def on_eof(): pass + def on_eof(): pass - def on_inference(msg): - # assistant and reasoning are assumed to have a Stream - # tool is assumed to have a ToolCall - if msg.role == 'assistant': - msg.content.listen(on_content, on_eof) - elif msg.role == 'reasoning': - msg.content.listen(on_reasoning, on_eof) - elif msg.role == 'tool': - deferred = route_tool_call(msg.content, tool_names) - msg_type = 'deferred_call' if deferred else 'tool_call' - loop.create_task(queue.put((msg_type, msg.content))) - elif msg.role == 'usage': - loop.create_task(queue.put(('usage', msg.content))) + async def drain(): + try: + async for msg in generator: + # assistant and reasoning are assumed to have a Stream + # tool is assumed to have a ToolCall + if msg.role == 'assistant': + msg.content.listen(on_content, on_eof) + elif msg.role == 'reasoning': + msg.content.listen(on_reasoning, on_eof) + elif msg.role == 'tool': + deferred = route_tool_call(msg.content, tool_names) + msg_type = 'deferred_call' if deferred else 'tool_call' + state['tool_call'] = True + await queue.put((msg_type, msg.content)) + elif msg.role == 'usage': + await queue.put(('usage', msg.content)) + except Exception as e: + # Surface the failure to the consumer via the eof drain. + state['error'] = e + finally: + await queue.put(('eof', None)) - def on_done(output): - loop.create_task(queue.put(('eof', None))) - - run.on('inference', on_inference) - run.on('done', on_done) - - return queue + loop.create_task(drain()) + return queue, state def format_chunk(agent_id, delta, finish_reason=None, usage=None): @@ -242,25 +247,23 @@ def register(server, app): # tool and gets deferred. tool_names = {spec.name for spec in tools} - # --- Get agent control and create AgentRun directly --- + # --- Get agent control and drive the run through the shared prompt generator --- 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, - # Never execute tools inside the run: emit them instead so - # the handler can send client tools back and defer native - # ones. - "tool_policy": "ignore", - }, + # The OpenAI inputs ride as kwargs so they land in run.input, which + # ExpertAgent.run reads ("tools", "tool_policy", "messages"). + generator = control.prompt( + None, + input=None, + messages=messages, + tools=tools, + # Never execute tools inside the run: emit them instead so + # the handler can send client tools back and defer native + # ones. + tool_policy="ignore", ) - queue = attach_run_handlers(run, tool_names) - await control.run(run) + queue, state = messages_to_queue(generator, tool_names) usage = {'completion_tokens': 0, 'prompt_tokens': 0, 'total_tokens': 0, 'cost': 0} @@ -303,10 +306,10 @@ def register(server, app): usage = data elif msg_type == 'eof': + if state['error']: + raise state['error'] # 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" + finish_reason = "tool_calls" if state['tool_call'] else "stop" yield f"data: {json.dumps(format_chunk(agent_id, {}, finish_reason))}\n\n" break @@ -334,6 +337,8 @@ def register(server, app): while True: msg_type, data = await queue.get() if msg_type == 'eof': + if state['error']: + raise state['error'] break elif msg_type == 'content': content += data diff --git a/backend/src/server/websocket.py b/backend/src/server/websocket.py index f1a2885..d3e1d3f 100644 --- a/backend/src/server/websocket.py +++ b/backend/src/server/websocket.py @@ -1,7 +1,6 @@ import json, asyncio, traceback from quart import websocket from utils import encode_json -from stream import Stream # isinstance check in send-agent-prompt def register(server, app, auth_manager): @server.websocket('/ws') @@ -298,37 +297,30 @@ def register(server, app, auth_manager): ) await send({'message': "chat thread created", 'threadId': thread_id, 'nonce': nonce}) - elif action == "send-agent-prompt": + elif action == "prompt-agent": if 'instruction' not in body and 'input' not in body: - await send({'error': "instruction or input required"}) + await send({'error': "instruction or input required", 'nonce': nonce}) continue - # Shape of response changes based on options lol - response = await app.send_agent_prompt( + # Multiple-result action, same shape as read-stream: every + # generated message is its own packet carrying the request + # nonce, so the client callback runs once per message. The + # terminal eof packet resolves the action. Streams referenced + # by open assistant/reasoning messages are read by the + # client via the existing read-stream action. + generator = app.prompt_agent( await agent_id_from_body(), body.get('instruction'), body.get('input'), body.get('options', {}) ) - print("prompt response", repr(response)) - tool_calls = None - - # Extract from dict - if isinstance(response, dict): - tool_calls = response['tool_calls'] - response = response['output'] - - if isinstance(response, Stream): - if body.get('stream'): - response = {"streamId": response.id} - else: - response = await response.content_wait() - - packet_out = {'message': "prompt sent", 'response': response, 'nonce': nonce} - if tool_calls: - packet_out['toolCalls'] = tool_calls - - await send(packet_out) + try: + async for msg in generator: + await send({'message': 'agent-message', 'data': msg, 'nonce': nonce}) + finally: + # Disconnect / cancellation point: closes the run task. + await generator.aclose() + await send({'eof': True, 'nonce': nonce}) elif action == "read-stream": stream_id = body['streamId'] diff --git a/frontend/src/views/Agent.vue b/frontend/src/views/Agent.vue index 607e49b..7b92981 100644 --- a/frontend/src/views/Agent.vue +++ b/frontend/src/views/Agent.vue @@ -157,14 +157,35 @@ } promptHistory.value.push(historyEntry); - let res = await backend.sendAction("send-agent-prompt", payload); - if (res.error) { - historyEntry.error = res.error; - } else { - historyEntry.output = res.response; + // prompt-agent is a multiple-result action (like read-stream): one packet + // per generated inference.Message, followed by a terminal eof packet. The + // callback runs once per packet. Assistant/reasoning content is an open + // stream ({streamId}); we read it live via read-stream. + historyEntry.output = ""; + try { + await backend.sendAction("prompt-agent", payload, (packet, deferred) => { + if (packet.eof) { + deferred.resolve(packet); + } else { + const msg = packet.data; + if (msg.role !== 'assistant') return; + const content = msg.content; + if (content && content.streamId != null) { + const stream = backend.readStream(content.streamId); + stream.on('data', () => { + historyEntry.output = stream.partialContent; + }); + } else if (typeof content === 'string') { + historyEntry.output += content; + } + } + }); + } catch (e) { + historyEntry.error = e.message || String(e); + } finally { + historyEntry.loading = false; } - historyEntry.loading = false; } function promptKeyDown (event, stage) {