File size: 4,242 Bytes
4a0cebd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | import threading
from abc import ABC
from typing import Dict, Optional
from loguru import logger
from pydantic import BaseModel, Field
from chat_engine.contexts.handler_context import HandlerContext
from chat_engine.data_models.chat_engine_config_data import ChatEngineConfigModel, HandlerBaseConfigModel
from chat_engine.common.handler_base import HandlerBase, HandlerBaseInfo, HandlerDataInfo, HandlerDetail
from chat_engine.data_models.chat_data.chat_data_model import ChatData
from chat_engine.data_models.chat_data_type import ChatDataType
from chat_engine.data_models.runtime_data.data_bundle import DataBundle, DataBundleDefinition, DataBundleEntry
from chat_engine.contexts.session_context import SessionContext
class GreeterConfig(HandlerBaseConfigModel, BaseModel):
greeting_prompt: str = Field(default="Introduce yourself to the visitor now, in your own words.")
delay_seconds: float = Field(default=4.0)
class GreeterContext(HandlerContext):
def __init__(self, session_id: str):
super().__init__(session_id)
self.timer = None
self.sent = False
class HandlerGreeter(HandlerBase, ABC):
"""On session start, injects one synthetic HUMAN_TEXT turn so the
avatar proactively greets the user before any mic input arrives."""
def __init__(self):
super().__init__()
self.greeting_prompt = "Introduce yourself to the visitor now, in your own words."
self.delay_seconds = 4.0
self.output_definition = None
def get_handler_info(self) -> HandlerBaseInfo:
return HandlerBaseInfo(
name="Greeter",
config_model=GreeterConfig,
)
def load(self, engine_config: ChatEngineConfigModel, handler_config: Optional[BaseModel] = None):
if isinstance(handler_config, GreeterConfig):
self.greeting_prompt = handler_config.greeting_prompt
self.delay_seconds = handler_config.delay_seconds
logger.info(f"Greeter loaded: delay={self.delay_seconds}s prompt={self.greeting_prompt!r}")
def get_handler_detail(self, session_context: SessionContext,
context: HandlerContext) -> HandlerDetail:
definition = DataBundleDefinition()
definition.add_entry(DataBundleEntry.create_audio_entry("avatar_audio", 1, 24000))
self.output_definition = definition
# AVATAR_TEXT input is declared only so the engine wires this handler
# into the graph; its data is ignored.
inputs = {
ChatDataType.AVATAR_TEXT: HandlerDataInfo(
type=ChatDataType.AVATAR_TEXT,
)
}
outputs = {
ChatDataType.HUMAN_TEXT: HandlerDataInfo(
type=ChatDataType.HUMAN_TEXT,
definition=definition,
)
}
return HandlerDetail(inputs=inputs, outputs=outputs)
def create_context(self, session_context, handler_config=None):
return GreeterContext(session_context.session_info.session_id)
def start_context(self, session_context, handler_context):
context = handler_context
if not isinstance(context, GreeterContext):
return
def _send_greeting():
if context.sent:
return
context.sent = True
try:
output = DataBundle(self.output_definition)
output.set_main_data(self.greeting_prompt)
context.submit_data(output, finish_stream=True)
logger.info(f"Greeter: injected greeting turn for session {context.session_id}")
except Exception as e:
logger.error(f"Greeter failed to inject greeting: {e}")
context.timer = threading.Timer(self.delay_seconds, _send_greeting)
context.timer.daemon = True
context.timer.start()
def handle(self, context: HandlerContext, inputs: ChatData,
output_definitions: Dict[ChatDataType, HandlerDataInfo]):
# Passive: greeting is emitted from start_context, inputs ignored.
return
def destroy_context(self, context: HandlerContext):
if isinstance(context, GreeterContext) and context.timer is not None:
context.timer.cancel()
|