usky-human/digitalHuman/server/core/api_agent_v0_impl.py

# -*- coding: utf-8 -*-


from typing import List, Dict
from digitalHuman.agent import AgentPool
from digitalHuman.utils import config
from digitalHuman.protocol import *
from digitalHuman.server.models import AgentEngineInput

agentPool = AgentPool()

def get_agent_list() -> List[EngineDesc]:
    agents = agentPool.list()
    return [agentPool.get(agent).desc() for agent in agents]

def get_agent_default() -> EngineDesc:
    return agentPool.get(config.SERVER.AGENTS.DEFAULT).desc()

def get_agent_param(name: str) -> List[ParamDesc]:
    engine = agentPool.get(name)
    return engine.parameters()

async def create_agent_conversation(name: str, param: Dict) -> str:
    engine = agentPool.get(name)
    id = await engine.createConversation(**param)
    return id

def agent_infer_stream(user: UserDesc, items: AgentEngineInput):
    # 检查是否是按钮触发的对话（包含 [BUTTON_TRIGGERED] 标记）
    # 如果是按钮触发，添加 persona 前缀；否则直接使用用户输入
    BUTTON_MARKER = "[BUTTON_TRIGGERED]"
    if items.data.startswith(BUTTON_MARKER):
        # 移除标记，添加 persona 前缀
        user_message = items.data[len(BUTTON_MARKER):]
        persona_prefix = "你现在是永天科技展厅的智能客服，请介绍永天科技的产品和解决方案：\n"
        user_input = persona_prefix + user_message
    else:
        # 普通对话，不添加 persona 前缀
        user_input = items.data
    
    input = TextMessage(data=user_input)
    streamContent = agentPool.get(items.engine).run(input=input, user=user, streaming=True, conversation_id=items.conversation_id, **items.config)
    return streamContent