揭秘高德地图智能助手：基于Assistant的优化实践与架构设计

@dataclass
class AppConfig:
    dashscope_api_key: str
    amap_api_key: str
    #... 其他配置项

"""基于 Assistant 实现的高德地图智能助手
优化版本改进：
1. 配置与代码分离
2. 更好的错误处理
3. 性能优化
4. 更安全的API密钥管理
5. 增强扩展性
"""
import os
import asyncio
from typing import Optional, Dict, Any
import logging
from dataclasses import dataclass
import dashscope
from qwen_agent.agents import Assistant
from qwen_agent.gui import WebUI

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

@dataclass
class AppConfig:
    """应用配置类"""
    dashscope_api_key: str
    amap_api_key: str
    llm_model: str = 'qwen-max'
    llm_timeout: int = 30
    llm_retry_count: int = 3
    mcp_server_command: str = 'npx'
    mcp_server_args: tuple = ('-y', '@amap/amap-maps-mcp-server')
    resource_root: str = os.path.join(os.path.dirname(__file__), 'resource')

class MapAssistant:
    """高德地图智能助手封装类"""
    def __init__(self, config: AppConfig):
        """初始化助手"""
        self.config = config
        self._validate_config()
        self._setup_environment()
        self.bot = self._init_assistant()

    def _validate_config(self):
        """验证配置"""
        if not self.config.dashscope_api_key:
            raise ValueError("DashScope API Key is required")
        if not self.config.amap_api_key:
            raise ValueError("AMap API Key is required")

    def _setup_environment(self):
        """设置环境变量"""
        os.environ['DASHSCOPE_API_KEY'] = self.config.dashscope_api_key
        dashscope.api_key = self.config.dashscope_api_key
    def _init_assistant(self) -> Assistant:
        """初始化助手实例"""
        llm_cfg = {
            'model': self.config.llm_model,
            'timeout': self.config.llm_timeout,
            'retry_count': self.config.llm_retry_count,
        }
        system_message = (
            '你是一个专业的地图助手，具有查询地图、规划路线、推荐景点等能力。'
            '你可以帮助用户规划旅游行程，查找地点，导航等。'
            '你应该充分利用高德地图的各种功能来提供专业的建议。'
            '回答要简洁明了，重点突出。'
        )

        tools = [{
            "mcpServers": {
                "amap-maps": {
                    "command": self.config.mcp_server_command,
                    "args": list(self.config.mcp_server_args),
                    "env": {
                        "AMAP_MAPS_API_KEY": self.config.amap_api_key
                    }
                }
            }
        }]

        try:
            return Assistant(
                llm=llm_cfg,
                name='智能地图助手',
                description='专业的地图查询与路线规划服务',
                system_message=system_message,
                function_list=tools,
            )
        except Exception as e:
            logger.error("助手初始化失败: %s", str(e))
            raise

    async def process_query(self, query: str, file: Optional[str] = None) -> Dict[str, Any]:
        """处理用户查询"""
        messages = []

        try:
            if not file:
                messages.append({'role': 'user', 'content': query})
            else:
                messages.append({'role': 'user', 'content': [{'text': query}, {'file': file}]})

            logger.info("Processing query: %s", query)
            response = []
            for resp in self.bot.run(messages):
                response.append(resp)

            return {
                'status': 'success',
                'response': response
            }
        except Exception as e:
            logger.error("处理查询时出错: %s", str(e))
            return {
                'status': 'error',
                'message': str(e)
            }

def load_config() -> AppConfig:
    """从环境变量加载配置"""
    return AppConfig(
        dashscope_api_key=os.getenv('DASHSCOPE_API_KEY'),
        amap_api_key=os.getenv('AMAP_API_KEY')
    )

def test(config: AppConfig, query: str = '帮我查找上海东方明珠的具体位置'):
    """测试模式"""
    try:
        assistant = MapAssistant(config)
        response = asyncio.run(assistant.process_query(query))
        print('Response:', response)
    except Exception as e:
        logger.error("测试失败: %s", str(e))

def app_gui(config: AppConfig):
    """启动Web界面"""
    try:
        assistant = MapAssistant(config)

        chatbot_config = {
            'prompt.suggestions': [
                '帮我规划上海一日游行程，主要想去外滩和迪士尼',
                '我在南京路步行街，帮我找一家评分高的本帮菜餐厅',
                '从浦东机场到外滩怎么走最方便？',
                '推荐上海三个适合拍照的网红景点',
                '帮我查找上海科技馆的具体地址和营业时间',
            ],
            'page.title': '智能地图助手',
            'page.icon': '🗺️'
        }

        WebUI(
            assistant.bot,
            chatbot_config=chatbot_config
        ).run()
    except Exception as e:
        logger.error("启动Web界面失败: %s", str(e))

if __name__ == '__main__':
    # 从环境变量加载配置
    config = load_config()

    # 运行模式选择
    # test(config)  # 测试模式
    app_gui(config)  # Web界面模式