chat-ai-api/app/utils/chat_utils.py

from typing import List, Optional, Dict
from ..schemas.chat import ChatMessage
from ..config.config import Config
from ..db.mongo import get_last_response_id


def convert_messages_for_api(messages: List[ChatMessage]) -> List[Dict[str, str]]:
    return [{"role": msg.role, "content": msg.content} for msg in messages]


def get_latest_user_message(messages: List[ChatMessage]) -> Optional[ChatMessage]:
    for message in reversed(messages):
        if message.role == "user":
            return message
    return None


def get_previous_response_id(user_id: str, session_id: str) -> Optional[str]:
    return get_last_response_id(user_id, session_id)

# 联网搜索工具
def get_web_search_tools() -> list:
    return  [{
        "type": "web_search",
        "max_keyword": 20,
        "limit": 20,
        "sources": ["douyin", "toutiao"],# 附加搜索来源（抖音百科、墨迹天气、头条图文等平台）
    }]
# 私域知识库 --- 此处的知识库也需要映射
def get_knowledge_search_tools(knowledge_resource_id: str, app_name: Optional[str] = None) -> list:
    tool: Dict = {
        "type": "knowledge_search",
        "knowledge_resource_id": knowledge_resource_id,  # 替换为实际知识库ID
        "limit": 10,  # 最多返回10条搜索结果
    }
    # 当 app_name 有值时，按 app_name 字段过滤，否则全局查询
    if app_name:
        tool["doc_filter"] = {
            "op": "must",
            "field": "app_name",
            "conds": [app_name],
        }
    return [tool]

# 豆包助手工具
def get_doubao_tools() -> list:
    return [{
        "type": "doubao_app",
        "feature": {
            # 联网搜索功能
            "ai_search": {
                "type": "disabled",
                "role_description": Config.ROLE_DESCRIPTION
            },
            # 边想边搜功能
            "reasoning_search":{
                "type": "enabled",
                "role_description": Config.ROLE_DESCRIPTION
            }
        }
    }]