from typing import List, Optional, Dict from ..schemas.chat import ChatMessage from ..core.ark_client 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", "moji", "toutiao"],# 附加搜索来源(抖音百科、墨迹天气、头条图文等平台) "user_location": { # 用户地理位置(用于优化搜索结果) "type": "approximate", # 大致位置 "country": "中国", "region": "浙江", "city": "杭州" } }] # 豆包助手工具 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 } }, "user_location": { "type": "approximate", "country": "中国", "region": "浙江", "city": "杭州" } }]