enh: channels streaming agent

Author: tjbck

backend/open_webui/main.py

@@ -1796,7 +1796,7 @@ async def chat_completion(
 
         if metadata.get('chat_id') and user:
             chat_id = metadata['chat_id']
-            if not chat_id.startswith('local:'):  # temporary chats are not stored
+            if not chat_id.startswith('local:') and not chat_id.startswith('channel:'):  # temporary/channel chats are not stored
                 if is_new_chat:
                     # Build the full history upfront with ALL assistant placeholders
                     user_message = metadata.get('user_message') or {}
@@ -2012,7 +2012,7 @@ async def emit_cancel_event():
             if metadata.get('chat_id') and metadata.get('message_id'):
                 # Update the chat message with the error
                 try:
-                    if not metadata['chat_id'].startswith('local:'):
+                    if not metadata['chat_id'].startswith('local:') and not metadata['chat_id'].startswith('channel:'):
                         await Chats.upsert_message_to_chat_by_id_and_message_id(
                             metadata['chat_id'],
                             metadata['message_id'],
@@ -2275,7 +2275,7 @@ async def list_tasks_endpoint(request: Request, user=Depends(get_admin_user)):
 
 @app.get('/api/tasks/chat/{chat_id:path}')
 async def list_tasks_by_chat_id_endpoint(request: Request, chat_id: str, user=Depends(get_verified_user)):
-    if chat_id.startswith('local:'):
+    if chat_id.startswith('local:') or chat_id.startswith('channel:'):
         socket_id = chat_id[len('local:') :]
         owner_id = get_user_id_from_session_pool(socket_id)
         if owner_id != user.id and user.role != 'admin':
@@ -2293,7 +2293,7 @@ async def list_tasks_by_chat_id_endpoint(request: Request, chat_id: str, user=De
 
 @app.post('/api/tasks/chat/{chat_id:path}/stop')
 async def stop_tasks_by_chat_id_endpoint(request: Request, chat_id: str, user=Depends(get_verified_user)):
-    if chat_id.startswith('local:'):
+    if chat_id.startswith('local:') or chat_id.startswith('channel:'):
         socket_id = chat_id[len('local:') :]
         owner_id = get_user_id_from_session_pool(socket_id)
         if owner_id != user.id and user.role != 'admin':

backend/open_webui/routers/channels.py

@@ -57,7 +57,7 @@
     get_all_models,
     get_filtered_models,
 )
-from open_webui.utils.chat import generate_chat_completion
+
 
 
 from open_webui.utils.auth import get_admin_user, get_verified_user
@@ -979,57 +979,50 @@ async def model_response_handler(request, channel, message, user, db=None):
                         ],
                     ]
 
+                # Resolve model config (same helpers automations use)
+                from open_webui.utils.automations import (
+                    _resolve_model_tool_ids,
+                    _resolve_model_features,
+                    _resolve_model_filter_ids,
+                )
+
+                tool_ids = _resolve_model_tool_ids(request.app, model_id)
+                features = _resolve_model_features(request.app, model_id)
+                filter_ids = _resolve_model_filter_ids(request.app, model_id)
+
+                # Build full form_data — same shape as frontend POST.
+                # The channel: prefix routes pipeline events to the
+                # channel emitter in socket/main.py instead of the
+                # default chat emitter.
                 form_data = {
                     'model': model_id,
                     'messages': [
                         system_message,
                         {'role': 'user', 'content': content},
                     ],
-                    'stream': False,
+                    'stream': True,
+                    'chat_id': f'channel:{channel.id}',
+                    'id': response_message.id,
+                    'session_id': f'channel:{channel.id}',
+                    'background_tasks': {},
                 }
-
-                res = await generate_chat_completion(
-                    request,
-                    form_data=form_data,
-                    user=user,
+                if tool_ids:
+                    form_data['tool_ids'] = tool_ids
+                if features:
+                    form_data['features'] = features
+                if filter_ids:
+                    form_data['filter_ids'] = filter_ids
+
+                # Call the full chat completion pipeline — streaming,
+                # tools, filters, RAG — everything. The pipeline runs as
+                # an async task; the channel emitter handles progressive
+                # message updates via socket events.
+                await request.app.state.CHAT_COMPLETION_HANDLER(
+                    request, form_data, user=user
                 )
 
-                if res:
-                    if res.get('choices', []) and len(res['choices']) > 0:
-                        await update_message_by_id(
-                            request,
-                            channel.id,
-                            response_message.id,
-                            MessageForm(
-                                **{
-                                    'content': res['choices'][0]['message']['content'],
-                                    'meta': {
-                                        'done': True,
-                                    },
-                                }
-                            ),
-                            user,
-                            db,
-                        )
-                    elif res.get('error', None):
-                        await update_message_by_id(
-                            request,
-                            channel.id,
-                            response_message.id,
-                            MessageForm(
-                                **{
-                                    'content': f'Error: {res["error"]}',
-                                    'meta': {
-                                        'done': True,
-                                    },
-                                }
-                            ),
-                            user,
-                            db,
-                        )
             except Exception as e:
-                log.info(e)
-                pass
+                log.exception(e)
 
     return True
 

backend/open_webui/socket/main.py

@@ -832,7 +832,80 @@ async def disconnect(sid):
         # print(f"Unknown session ID {sid} disconnected")
 
 
+async def _make_channel_emitter(request_info):
+    """Event emitter that routes pipeline output to a channel message.
+
+    Translates chat:completion events into channel message:update socket
+    emissions, throttled to avoid flooding with per-token updates.
+    """
+    channel_id = request_info['chat_id'].removeprefix('channel:')
+    message_id = request_info['message_id']
+
+    state = {'last_emit_at': 0.0}
+    THROTTLE_INTERVAL = 0.15  # ~6 updates/sec
+
+    async def _emit_channel_update(content: str, done: bool = False):
+        from open_webui.models.messages import Messages, MessageForm
+
+        update_form = MessageForm(content=content)
+        if done:
+            # Merge done flag into existing meta (preserve model_id etc.)
+            msg = await Messages.get_message_by_id(message_id)
+            existing_meta = (msg.meta or {}) if msg else {}
+            update_form = MessageForm(
+                content=content,
+                meta={**existing_meta, 'done': True},
+            )
+
+        await Messages.update_message_by_id(message_id, update_form)
+        message = await Messages.get_message_by_id(message_id)
+        if message:
+            await sio.emit(
+                'events:channel',
+                {
+                    'channel_id': channel_id,
+                    'message_id': message_id,
+                    'data': {
+                        'type': 'message:update',
+                        'data': message.model_dump(),
+                    },
+                },
+                to=f'channel:{channel_id}',
+            )
+
+    async def __channel_emitter__(event_data):
+        event_type = event_data.get('type')
+
+        if event_type == 'chat:completion':
+            data = event_data.get('data', {})
+            content = data.get('content', '')
+            done = data.get('done', False)
+
+            if not content and not done:
+                return
+
+            now = __import__('time').time()
+            if done or (now - state['last_emit_at']) >= THROTTLE_INTERVAL:
+                state['last_emit_at'] = now
+                await _emit_channel_update(content, done)
+
+        elif event_type == 'chat:message:error':
+            error = event_data.get('data', {}).get('error', {})
+            error_content = (
+                error.get('content', 'An error occurred')
+                if isinstance(error, dict)
+                else str(error)
+            )
+            await _emit_channel_update(f'Error: {error_content}', done=True)
+
+    return __channel_emitter__
+
+
 async def get_event_emitter(request_info, update_db=True):
+    # Channel mode: route pipeline output to channel message updates
+    if request_info.get('chat_id', '').startswith('channel:'):
+        return await _make_channel_emitter(request_info)
+
     async def __event_emitter__(event_data):
         user_id = request_info['user_id']
         chat_id = request_info['chat_id']