wehub-resource-sync
e4dcfc49aa
Tests / Import Check (Python 3.13) (push) Has been cancelled
Tests / Import Check (Python 3.14) (push) Has been cancelled
Tests / Python Tests (Python 3.11) (push) Has been cancelled
Tests / Python Tests (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.14) (push) Has been cancelled
Tests / Test Summary (push) Has been cancelled
Tests / Lint and Format (push) Has been cancelled
Tests / Web Node Tests (push) Has been cancelled
Tests / Import Check (Python 3.11) (push) Has been cancelled
Tests / Import Check (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.13) (push) Has been cancelled
127 lines
4.3 KiB
Python
127 lines
4.3 KiB
Python
"""
|
|
Chat Orchestrator
|
|
=================
|
|
|
|
Unified entry point that routes user messages to the appropriate capability.
|
|
All consumers (CLI, WebSocket, SDK) call the orchestrator.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from typing import Any, AsyncIterator
|
|
import uuid
|
|
|
|
from deeptutor.core.context import UnifiedContext
|
|
from deeptutor.core.stream import StreamEvent, StreamEventType
|
|
from deeptutor.core.stream_bus import StreamBus, register_bus, unregister_bus
|
|
from deeptutor.events.event_bus import Event, EventType, get_event_bus
|
|
from deeptutor.runtime.registry.capability_registry import get_capability_registry
|
|
from deeptutor.runtime.registry.tool_registry import get_tool_registry
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ChatOrchestrator:
|
|
"""
|
|
Routes a ``UnifiedContext`` to the correct capability, manages
|
|
the ``StreamBus`` lifecycle, and publishes completion events.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._cap_registry = get_capability_registry()
|
|
self._tool_registry = get_tool_registry()
|
|
|
|
async def handle(self, context: UnifiedContext) -> AsyncIterator[StreamEvent]:
|
|
"""
|
|
Execute a single user turn and yield streaming events.
|
|
|
|
If ``context.active_capability`` is set, the corresponding capability
|
|
handles the turn. Otherwise, the default ``chat`` capability is used.
|
|
"""
|
|
if not context.session_id:
|
|
context.session_id = str(uuid.uuid4())
|
|
|
|
cap_name = context.active_capability or "chat"
|
|
capability = self._cap_registry.get(cap_name)
|
|
|
|
if capability is None:
|
|
bus = StreamBus()
|
|
await bus.error(
|
|
f"Unknown capability: {cap_name}. "
|
|
f"Available: {self._cap_registry.list_capabilities()}",
|
|
source="orchestrator",
|
|
)
|
|
await bus.close()
|
|
async for event in bus.subscribe():
|
|
yield event
|
|
return
|
|
|
|
yield StreamEvent(
|
|
type=StreamEventType.SESSION,
|
|
source="orchestrator",
|
|
metadata={
|
|
"session_id": context.session_id,
|
|
"turn_id": str(context.metadata.get("turn_id", "")),
|
|
},
|
|
)
|
|
|
|
bus = StreamBus()
|
|
_turn_id = str(context.metadata.get("turn_id") or "")
|
|
if _turn_id:
|
|
register_bus(_turn_id, bus)
|
|
|
|
async def _run() -> None:
|
|
try:
|
|
await capability.run(context, bus)
|
|
except Exception as exc:
|
|
logger.error("Capability %s failed: %s", cap_name, exc, exc_info=True)
|
|
await bus.error(str(exc), source=cap_name)
|
|
finally:
|
|
await bus.emit(StreamEvent(type=StreamEventType.DONE, source=cap_name))
|
|
await bus.close()
|
|
if _turn_id:
|
|
unregister_bus(_turn_id)
|
|
|
|
stream = bus.subscribe()
|
|
task = asyncio.create_task(_run())
|
|
|
|
async for event in stream:
|
|
yield event
|
|
|
|
await task
|
|
await self._publish_completion(context, cap_name)
|
|
|
|
async def _publish_completion(self, context: UnifiedContext, cap_name: str) -> None:
|
|
"""Publish CAPABILITY_COMPLETE to the global EventBus."""
|
|
try:
|
|
bus = get_event_bus()
|
|
await bus.publish(
|
|
Event(
|
|
type=EventType.CAPABILITY_COMPLETE,
|
|
task_id=str(context.metadata.get("turn_id") or context.session_id),
|
|
user_input=context.user_message,
|
|
agent_output="",
|
|
metadata={
|
|
"capability": cap_name,
|
|
"session_id": context.session_id,
|
|
"turn_id": str(context.metadata.get("turn_id", "")),
|
|
},
|
|
)
|
|
)
|
|
except Exception:
|
|
logger.debug("EventBus publish failed (may not be running)", exc_info=True)
|
|
|
|
def list_tools(self) -> list[str]:
|
|
return self._tool_registry.list_tools()
|
|
|
|
def list_capabilities(self) -> list[str]:
|
|
return self._cap_registry.list_capabilities()
|
|
|
|
def get_capability_manifests(self) -> list[dict[str, Any]]:
|
|
return self._cap_registry.get_manifests()
|
|
|
|
def get_tool_schemas(self, names: list[str] | None = None) -> list[dict[str, Any]]:
|
|
return self._tool_registry.build_openai_schemas(names)
|