解决方案设计

方案概述

基于对 Kimi CLI web_search 实现的深入分析，我们提出两套可行方案：

• 方案 A: 基于 Moonshot AI 生态的集成方案（需等待官方开放搜索 API）
• 方案 B: 基于第三方搜索 API 的变通方案（立即可实施）

方案 A: Moonshot AI 原生集成（长期规划）

适用场景

当你希望深度集成 Moonshot AI 生态，并且：

• 已有 Moonshot AI 开放平台账户
• 未来官方可能开放搜索能力
• 需要与 Kimi 模型深度协同

架构设计

graph TD
    A[用户查询] --> B[SearchWeb Tool]
    B --> C{OAuth认证}
    C -->|已登录| D[调用 api.kimi.com/coding/v1/search]
    C -->|未登录| E[引导登录]
    D --> F[返回搜索结果]
    F --> G[格式化输出]
    G --> H[AI消费]

技术实现要点

1. 认证流程

# 伪代码示例
async def search_with_moonshot(query: str):
    # 1. 获取访问令牌
    token = await oauth_manager.get_token()
    
    # 2. 构造请求
    headers = {
        "Authorization": f"Bearer {token}",
        "X-Msh-Platform": "your-app",
    }
    
    # 3. 调用 API
    response = await http_client.post(
        "https://api.kimi.com/coding/v1/search",
        headers=headers,
        json={
            "text_query": query,
            "limit": 5,
            "enable_page_crawling": False,
        }
    )
    
    return response.json()

2. 前置条件

要使用此方案，你需要：

方案 A 评估

结论: 目前不可行，建议持续关注 Moonshot AI 开放平台更新。

方案 B: 第三方搜索 API 变通方案（推荐）

适用场景

当你需要立即实现搜索功能，并且：

• 希望快速上线
• 可以接受非 Moonshot 官方搜索能力
• 有一定预算用于搜索 API 调用

可选第三方服务对比

推荐架构: 以 Bing Search API 为例

graph LR
    A[用户查询] --> B[Search Tool]
    B --> C[配置管理]
    C --> D[Bing Search API]
    D --> E[结果解析]
    E --> F[格式化为 Kimi CLI 风格]
    F --> G[AI消费]
    
    subgraph 配置管理
    H[API Key]
    I[Endpoint]
    J[参数]
    end

技术实现

1. 配置定义（借鉴 Kimi CLI 风格）

from pydantic import BaseModel, SecretStr
from typing import Literal

class SearchConfig(BaseModel):
    """搜索服务配置"""
    provider: Literal["bing", "google", "exa"]
    api_key: SecretStr
    base_url: str  # 例如: https://api.bing.microsoft.com/v7.0
    custom_headers: dict[str, str] | None = None
    default_limit: int = 5
    timeout_seconds: int = 30

2. 搜索工具实现

import aiohttp
from typing import List, Dict, Any

class WebSearchTool:
    """Web 搜索工具 - 兼容 Kimi CLI 风格"""
    
    def __init__(self, config: SearchConfig):
        self.config = config
        self.session = None
    
    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()
    
    async def search(
        self, 
        query: str, 
        limit: int = 5,
        include_content: bool = False
    ) -> List[Dict[str, Any]]:
        """
        执行网页搜索
        
        Args:
            query: 搜索关键词
            limit: 返回结果数量 (1-20)
            include_content: 是否抓取页面全文（需额外处理）
        
        Returns:
            格式化的搜索结果列表
        """
        if self.config.provider == "bing":
            return await self._search_bing(query, limit)
        elif self.config.provider == "exa":
            return await self._search_exa(query, limit)
        else:
            raise ValueError(f"Unsupported provider: {self.config.provider}")
    
    async def _search_bing(self, query: str, limit: int) -> List[Dict[str, Any]]:
        """Bing Search API 实现"""
        headers = {
            "Ocp-Apim-Subscription-Key": self.config.api_key.get_secret_value(),
        }
        
        params = {
            "q": query,
            "count": min(limit, 50),  # Bing 最大 50
            "mkt": "zh-CN",  # 中文市场
        }
        
        async with self.session.get(
            f"{self.config.base_url}/search",
            headers=headers,
            params=params
        ) as response:
            data = await response.json()
            
            # 转换为 Kimi CLI 风格的格式
            results = []
            for item in data.get("webPages", {}).get("value", []):
                results.append({
                    "site_name": item.get("siteName", ""),
                    "title": item.get("name", ""),
                    "url": item.get("url", ""),
                    "snippet": item.get("snippet", ""),
                    "date": item.get("dateLastCrawled", ""),
                    "content": "",  # 需要额外抓取
                })
            
            return results
    
    async def _search_exa(self, query: str, limit: int) -> List[Dict[str, Any]]:
        """Exa.ai API 实现"""
        headers = {
            "Authorization": f"Bearer {self.config.api_key.get_secret_value()}",
            "Content-Type": "application/json",
        }
        
        payload = {
            "query": query,
            "numResults": limit,
            "useAutoprompt": True,
            "type": "auto",
        }
        
        async with self.session.post(
            "https://api.exa.ai/search",
            headers=headers,
            json=payload
        ) as response:
            data = await response.json()
            
            results = []
            for item in data.get("results", []):
                results.append({
                    "site_name": item.get("title", "").split(" - ")[-1],
                    "title": item.get("title", ""),
                    "url": item.get("url", ""),
                    "snippet": item.get("snippet", ""),
                    "date": "",
                    "content": item.get("text", ""),
                })
            
            return results

3. 结果格式化（完全兼容 Kimi CLI 格式）

def format_results(results: List[Dict[str, Any]]) -> str:
    """将搜索结果格式化为 Kimi CLI 风格"""
    output = []
    
    for i, result in enumerate(results):
        if i > 0:
            output.append("---\n")
        
        output.append(f"Title: {result['title']}")
        output.append(f"Date: {result.get('date', '')}")
        output.append(f"URL: {result['url']}")
        output.append(f"Summary: {result['snippet']}\n")
        
        if result.get('content'):
            output.append(f"{result['content']}\n")
    
    return "\n".join(output)

方案 B 评估

结论: 推荐方案，可快速实施且成本可控。

方案对比矩阵

架构设计建议

无论你选择哪个方案，都建议采用以下架构模式（借鉴 Kimi CLI 的优秀设计）：

1. 工具化抽象

class SearchTool:
    """搜索工具基类"""
    name: str = "WebSearch"
    description: str = "搜索互联网获取最新信息"
    
    async def execute(self, query: str, **kwargs) -> str:
        """执行搜索并返回格式化结果"""
        raise NotImplementedError

2. 配置外部化

• 敏感信息（API Key）使用环境变量或密钥管理器
• 服务端点可配置，便于切换服务商
• 参数默认值可覆盖

3. 错误处理策略

try:
    results = await search_tool.search(query)
except SearchAPIError as e:
    return f"搜索服务暂时不可用: {e}"
except RateLimitError:
    return "搜索频率超限，请稍后再试"
except Exception as e:
    return f"搜索失败: {e}"

4. 缓存机制（可选优化）

对相同查询结果进行短期缓存，减少 API 调用成本：

from functools import lru_cache

@lru_cache(maxsize=100)
async def cached_search(query: str, limit: int = 5):
    return await search_tool.search(query, limit)

最终建议

短期（1-2 周）

采用方案 B，使用 Bing Search API 或 Exa.ai 快速实现搜索功能：

1. 注册 Bing Search API（免费额度 1000 次/月）
2. 实现 SearchTool 类
3. 集成到你的 AI Agent 中

长期（3-6 个月）

持续关注 Moonshot AI 开放平台动态：

1. 订阅官方公告
2. 当搜索 API 开放时，可平滑迁移到方案 A
3. 架构设计保持服务商无关性，便于切换

参考资料

• Microsoft Bing Search API 文档
• Exa.ai API 文档
• Google Custom Search JSON API