web-search-agent

第1章 · Agent 基础知识 · 配套项目 chapter1/web-search-agent

项目说明

Kimi Web Search Agent 🔍

一个基于 Kimi API 的智能搜索 Agent，能够理解用户问题，自动进行网络搜索并总结答案。

📋 项目概述

本项目实现了一个自主式 AI Agent，利用 Kimi（Moonshot AI）的内置 Web 搜索工具（search 和 crawl），能够：

• 🤔 智能理解：分析用户问题，识别关键信息需求
• 🔍 自动搜索：使用 Kimi 的内置 $web_search 工具获取实时网络信息
• 🔄 迭代搜索：可以多次调用搜索工具获取更全面的信息
• 📝 智能总结：综合多源信息，生成准确、全面的答案

🏗️ 架构设计

graph TD
    A[用户问题] --> B{Agent 思考}
    B -->|需要搜索| C[调用 $web_search 工具]
    C --> D[Kimi 搜索引擎]
    D --> E[返回搜索结果]
    E --> F{信息充足?}
    F -->|否| G[继续调用 $web_search]
    G --> H[获取更多信息]
    H --> F
    F -->|是| I[生成最终答案]
    B -->|不需要搜索| J[直接回答]

🚀 快速开始

1. 安装依赖

pip install -r requirements.txt

2. 配置 API Key

从 Moonshot AI 平台 获取 API Key，然后设置环境变量：

export MOONSHOT_API_KEY='your-api-key-here'

或创建 .env 文件：

MOONSHOT_API_KEY=your-api-key-here

注意: 为了向后兼容，系统也支持使用 KIMI_API_KEY 环境变量。

通用兜底（OpenRouter）: 若未设置 MOONSHOT_API_KEY/KIMI_API_KEY 但设置了 OPENROUTER_API_KEY，请求会自动改走 OpenRouter，使用 OPENROUTER_MODEL（默认 openai/gpt-5.6-luna）。重要限制：Kimi 内置的 $web_search 工具是 Moonshot 专有能力， 在 OpenRouter 上不可用——因此兜底模式下模型仅凭自身知识作答，没有实时联网搜索。 如需真正的联网搜索，请使用 Moonshot 主 key。

3. 运行 Agent

main.py 提供了完整的命令行接口（中文帮助）。查看全部参数：

python main.py --help

参数	说明	默认值
query	要提问的问题（位置参数）；省略则进入交互模式	无
--provider	搜索后端：kimi（调用内置 $web_search，需 API Key）/ offline-demo（离线示例轨迹）	kimi
--model	模型名称	kimi-k3
--max-steps	最大 ReAct 迭代次数	5
--base-url	API 基础 URL	https://api.moonshot.cn/v1
--api-key	Kimi API Key（默认读环境变量）	环境变量
--output, -o	将问题、ReAct 轨迹与答案保存为 JSON	无
--quiet	不实时打印 ReAct 轨迹	打印

离线演示 ReAct 循环（无需 API Key，回放示例轨迹，直观展示“想→做→看”）：

python main.py --provider offline-demo

交互模式（持续对话）：

python main.py

单次问答（运行时逐步打印思考/行动/观察轨迹）：

python main.py "2024年诺贝尔物理学奖获得者是谁？"
python main.py "比特币现价" --max-steps 3 --output result.json

快速体验（引导式交互）：

python quickstart.py

高级示例：

python examples.py

💡 运行时会实时打印 ReAct 轨迹：💭 思考 → 🔧 行动（调用 $web_search）→ 👀 观察（搜索结果）→ ✅ 最终答案，对应本章讲的“想→做→看”循环。agent.get_trace() 可获取结构化轨迹，--output 可将其存为 JSON。

📖 使用示例

基础使用

from agent import WebSearchAgent
from config import Config

# 创建 Agent
agent = WebSearchAgent(api_key=Config.get_api_key())

# 提问并获取答案
question = "Python 3.12 有哪些新特性？"
answer = agent.search_and_answer(question)
print(answer)

高级功能

运行高级示例：

python examples.py

包含以下功能： - 📦 批量搜索：同时搜索多个问题 - 🎯 带上下文搜索：提供背景信息进行更精准的搜索 - ⚖️ 比较搜索：搜索并比较多个项目 - ✅ 事实核查：验证陈述的真实性 - 📚 研究助手：深度研究某个主题

🛠️ 核心组件

agent.py - 核心 Agent 实现

• WebSearchAgent: 主要的 Agent 类
• search_and_answer(): 执行 ReAct 循环并生成答案的主方法
• get_trace(): 返回上一次运行的结构化 ReAct 轨迹（思考/行动/观察/最终答案）
• _chat(): 与 Kimi API 进行对话交互
• _get_system_prompt(): 获取系统提示，定义 Agent 行为
• _get_tools(): 定义可用的工具（$web_search）
• search_impl(): 搜索实现的抽象层，便于扩展
• format_trace_step(): 将一条轨迹步骤渲染为可读文本
• run_offline_demo(): 离线回放示例轨迹，无需 API Key 即可演示 ReAct 循环

config.py - 配置管理

• API 配置
• 模型选择
• 搜索参数设置

main.py - 主程序入口

• build_parser(): argparse 命令行接口（中文帮助，见 --help）
• run_interactive_mode(): 交互式对话模式
• run_single_question(): 单次问答模式
• 离线演示模式（--provider offline-demo）与 JSON 结果输出（--output）
• 会话管理

quickstart.py - 快速体验脚本

• demo_search(): 演示搜索功能
• interactive_mode(): 简化的交互模式
• 彩色输出和用户引导
• API Key 配置检查

examples.py - 高级示例

• AdvancedWebSearchAgent: 扩展功能的 Agent 类
• batch_search(): 批量处理多个问题
• search_with_context(): 带上下文的搜索
• comparative_search(): 比较多个项目
• fact_check(): 事实验证功能
• example_research_assistant(): 深度研究示例

🔧 配置选项

配置项	说明	默认值
MOONSHOT_API_KEY	Moonshot AI API 密钥	必填
KIMI_API_KEY	旧版 API 密钥变量名（向后兼容）	可选
KIMI_BASE_URL	API 基础 URL	https://api.moonshot.cn/v1
DEFAULT_MODEL	默认模型	kimi-k3
MAX_SEARCH_ITERATIONS	最大搜索迭代次数（Config 中设置）	5
SEARCH_TIMEOUT	搜索超时时间（秒）	30
temperature	控制生成内容的创造性	0.6

📊 技术特点

核心技术

• Kimi API: 使用 Moonshot AI 的最新 Kimi K3 模型（kimi-k3，原生联网搜索的推理模型）
• 内置工具调用: 利用 Kimi 的 $web_search 内置函数
• 迭代式搜索: 支持多轮搜索直到获得充分信息（最多5次迭代）
• 上下文管理: 维护完整对话历史，支持连续对话
• 温度控制: 支持调整生成内容的创造性（temperature 参数）

优势

• ✅ 实时信息: 获取最新的网络信息
• ✅ 智能理解: 理解用户意图，精准搜索
• ✅ 结构化输出: 生成组织良好的答案
• ✅ 可扩展性: 易于添加新功能和工具

📝 开发计划

• [ ] 添加异步搜索支持（使用 aiohttp）
• [ ] 实现搜索结果缓存机制
• [ ] 支持更多搜索后端（通过 search_impl 扩展）
• [ ] 支持多语言搜索
• [ ] 添加搜索结果质量评分
• [ ] 实现搜索历史记录
• [ ] 集成重试机制（使用 tenacity）
• [ ] 优化长对话的上下文管理

📄 许可证

MIT License

🔗 相关链接

• Kimi API 文档
• Web 搜索工具文档
• Moonshot AI 平台

⚠️ 注意事项

1. API 限制: 请注意 Kimi API 的调用限制和配额
2. 搜索质量: 搜索结果质量依赖于 Kimi 的搜索能力
3. 响应时间: 网络搜索可能需要一定时间，请耐心等待
4. 内容准确性: Agent 会尽力提供准确信息，但建议对重要信息进行二次验证

💡 使用建议

1. 明确问题: 提供清晰、具体的问题以获得更好的答案
2. 提供上下文: 必要时提供背景信息帮助 Agent 理解
3. 迭代优化: 如果答案不满意，可以提供更多细节重新提问
4. 合理期望: Agent 基于搜索结果回答，可能无法回答所有问题

---

作者: AI Agent 实战训练营
版本: 1.0.0
更新时间: 2024

源代码

agent.py

"""
Kimi Web Search Agent
一个基于 Kimi API 的智能搜索 Agent，能够理解用户问题，通过搜索引擎获取信息，并总结出答案。
"""

import json
from typing import List, Dict, Any, Optional
from openai import OpenAI
from openai.types.chat.chat_completion import Choice
import logging
import os

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


def _reasoning_safe_temperature(model, requested=1.0):
    """Reasoning models (Kimi K3, GPT-5, ...) only accept temperature=1.
    Return 1 for those; otherwise the requested value so non-reasoning
    providers (Doubao, DeepSeek, older Moonshot) are unchanged."""
    m = str(model or "").lower().replace("/", "-")
    return 1 if ("kimi-k3" in m or "gpt-5" in m) else requested


# ReAct 轨迹的步骤类型与展示标签（思考 → 行动 → 观察 → 最终答案）
STEP_LABELS = {
    "thought": ("💭", "思考"),
    "action": ("🔧", "行动"),
    "observation": ("👀", "观察"),
    "answer": ("✅", "最终答案"),
}


def format_trace_step(step: Dict[str, Any], max_len: int = 500) -> str:
    """把一条 ReAct 轨迹步骤渲染成一行可读文本。

    这正是本章强调的“轨迹（trajectory）”——用户消息、模型思考、工具调用、
    工具结果都被清晰地区分开来，让 ReAct 循环“想→做→看”一目了然。
    """
    icon, label = STEP_LABELS.get(step["type"], ("•", step["type"]))
    prefix = f"{icon} [{step.get('iteration', '-')}] {label}"

    if step["type"] == "action":
        args = json.dumps(step.get("args", {}), ensure_ascii=False)
        return f"{prefix}: 调用工具 {step.get('tool')}  参数={args}"

    content = str(step.get("content", "")).strip()
    if len(content) > max_len:
        content = content[:max_len] + f"…（省略 {len(content) - max_len} 字）"
    return f"{prefix}: {content}"


def search_impl(arguments: Dict[str, Any]) -> Any:
    """
    When using the search tool provided by Moonshot AI, you just need to return the arguments as they are,
    without any additional processing logic.

    But if you want to use other models and keep the internet search functionality, you just need to modify 
    the implementation here (for example, calling search and fetching web page content), the function signature 
    remains the same and still works.

    This ensures maximum compatibility, allowing you to switch between different models without making 
    destructive changes to the code.
    """
    return arguments


class WebSearchAgent:
    """
    Web Search Agent - 使用 Kimi API 的内置搜索工具

    根据官方文档: https://platform.moonshot.ai/docs/guide/use-web-search
    Kimi 提供了内置的 $web_search 工具
    """

    def __init__(self, api_key: str = None, base_url: str = "https://api.moonshot.cn/v1",
                 model: str = "kimi-k3", verbose: bool = False):
        """
        初始化 Agent

        Args:
            api_key: Kimi API key (如果不提供，从环境变量获取)
            base_url: API 基础 URL
            model: 使用的模型名称（默认 kimi-k3）
            verbose: 是否实时打印 ReAct 轨迹（思考/行动/观察）
        """
        # 优先使用传入的 api_key，否则从环境变量获取
        # Moonshot 为主，OpenRouter 为通用兜底（当 MOONSHOT_API_KEY 缺失时启用）
        from config import resolve_llm_backend
        primary_key = api_key or os.environ.get("MOONSHOT_API_KEY") or os.environ.get("KIMI_API_KEY")
        resolved_key, resolved_base_url, model, self.using_openrouter = \
            resolve_llm_backend(primary_key, base_url, model)
        if self.using_openrouter:
            logger.info(
                f"MOONSHOT_API_KEY 未设置，改用 OpenRouter 兜底（模型: {model}）。"
                "注意：Moonshot 内置 $web_search 工具在 OpenRouter 上不可用，"
                "此模式下模型将仅凭自身知识作答，不做实时联网搜索。"
            )

        self.client = OpenAI(
            api_key=resolved_key,
            base_url=resolved_base_url
        )
        self.model = model
        self.verbose = verbose
        self.conversation_history = []
        # ReAct 轨迹：按顺序记录每一步的思考/行动/观察，便于展示与调试
        self.trace: List[Dict[str, Any]] = []
        self.temperature = 0.6
        # 推理模型（Kimi K3）需要充足的输出预算，避免最终答案被截断
        self.max_tokens = 4096

    def _emit(self, step: Dict[str, Any]):
        """记录一条 ReAct 轨迹步骤，并在 verbose 模式下实时打印。"""
        self.trace.append(step)
        if self.verbose:
            print(format_trace_step(step))

    def _get_tools(self) -> List[Dict[str, Any]]:
        """
        定义可用的工具
        根据 Kimi 文档，$web_search 是内置工具（仅 Moonshot 支持）。
        经 OpenRouter 兜底时该内置工具不可用，返回空列表以避免 400 错误，
        此时模型仅凭自身知识作答。
        """
        if getattr(self, "using_openrouter", False):
            return []
        return [
            {
                "type": "builtin_function",
                "function": {
                    "name": "$web_search",
                }
            }
        ]

    def _get_system_prompt(self) -> str:
        """
        获取系统提示
        """
        return f"""你是 Kimi，一个智能搜索助手。

请按照以下步骤处理：
1. 分析用户问题，识别关键信息需求
2. 使用 $web_search 工具搜索相关信息
3. 如果需要更多信息，可以多次调用搜索工具
4. 综合所有信息，生成准确、全面的答案

注意：
- 搜索时使用精准的关键词
- 优先获取最新、最权威的信息
- 答案要结构清晰，有理有据
"""

    def _chat(self, messages: List[Dict[str, Any]]) -> Choice:
        """
        调用 Kimi API 进行对话

        Args:
            messages: 消息列表

        Returns:
            API 响应的 Choice 对象
        """
        kwargs = dict(
            model=self.model,
            messages=messages,
            temperature=_reasoning_safe_temperature(self.model, self.temperature),
            # Kimi K3 是推理模型，会先产出较长的 reasoning_content，需要给最终回答
            # 留足输出预算（Moonshot 要求 max_tokens>=2048），否则答案可能被截断为空。
            max_tokens=self.max_tokens,
        )
        tools = self._get_tools()
        if tools:  # OpenRouter 兜底时无内置搜索工具，省略 tools 参数
            kwargs["tools"] = tools
        completion = self.client.chat.completions.create(**kwargs)
        return completion.choices[0]

    def search_and_answer(self, user_question: str, max_iterations: int = 5) -> str:
        """
        执行搜索并生成答案

        Args:
            user_question: 用户问题
            max_iterations: 最大搜索迭代次数（防止无限循环）

        Returns:
            最终答案
        """
        # 构建系统提示
        system_prompt = self._get_system_prompt()

        # 重置对话历史并添加新的系统提示
        self.conversation_history = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_question}
        ]
        # 重置 ReAct 轨迹
        self.trace = []
        logger.info("开始调用 Kimi 搜索工具...")

        try:
            finish_reason = None
            iteration = 0

            # 循环处理，直到获得最终答案或达到最大迭代次数
            while (finish_reason is None or finish_reason == "tool_calls") and iteration < max_iterations:
                iteration += 1
                logger.info(f"迭代 {iteration}/{max_iterations}")

                # 调用 Kimi API
                choice = self._chat(self.conversation_history)
                finish_reason = choice.finish_reason

                # 捕获模型的思考过程（Kimi K3 等推理模型通过 reasoning_content 暴露思考模式）
                reasoning = getattr(choice.message, "reasoning_content", None)
                if reasoning:
                    self._emit({"iteration": iteration, "type": "thought", "content": reasoning})

                if finish_reason == "tool_calls":
                    # 处理工具调用
                    logger.info(f"模型请求调用 {len(choice.message.tool_calls)} 个工具")

                    # 添加助手的消息（包含工具调用）到历史。
                    # 注意：必须把消息重建为纯 dict，而不是直接塞入 SDK 返回的
                    # pydantic message 对象——后者会附带 reasoning_content / refusal
                    # 等额外字段，回传给 Moonshot 时会触发 "tokenization failed" 400 错误。
                    self.conversation_history.append({
                        "role": "assistant",
                        "content": choice.message.content or "",
                        "tool_calls": [
                            {
                                "id": tc.id,
                                "type": "function",
                                "function": {
                                    "name": tc.function.name,
                                    "arguments": tc.function.arguments,
                                },
                            }
                            for tc in choice.message.tool_calls
                        ],
                    })

                    # 执行每个工具调用
                    for tool_call in choice.message.tool_calls:
                        tool_call_name = tool_call.function.name
                        tool_call_arguments = json.loads(tool_call.function.arguments)

                        logger.info(f"执行工具: {tool_call_name}, 参数: {tool_call_arguments}")
                        # 行动：记录一次工具调用
                        self._emit({"iteration": iteration, "type": "action",
                                    "tool": tool_call_name, "args": tool_call_arguments})

                        if tool_call_name == "$web_search":
                            # 调用搜索实现
                            tool_result = search_impl(tool_call_arguments)
                        else:
                            tool_result = f"Error: unable to find tool by name '{tool_call_name}'"

                        tool_content = json.dumps(tool_result, ensure_ascii=False)
                        # 观察：记录工具返回结果
                        self._emit({"iteration": iteration, "type": "observation",
                                    "tool": tool_call_name, "content": tool_content})
                        # 构建工具响应消息并添加到历史
                        self.conversation_history.append({
                            "role": "tool",
                            "tool_call_id": tool_call.id,
                            "name": tool_call_name,
                            "content": tool_content
                        })
                else:
                    # 获得最终答案
                    if choice.message.content:
                        answer = choice.message.content
                        logger.info("成功生成答案")
                        self._emit({"iteration": iteration, "type": "answer", "content": answer})

                        # 添加最终答案到历史
                        self.conversation_history.append({
                            "role": "assistant",
                            "content": answer
                        })

                        return answer

            # 如果达到最大迭代次数仍未完成
            if iteration >= max_iterations:
                logger.warning(f"达到最大迭代次数 {max_iterations}")
                return "抱歉，搜索过程超过了最大迭代次数，请稍后重试。"

            return "抱歉，我无法获取足够的信息来回答您的问题。"

        except Exception as e:
            logger.error(f"搜索过程中出现错误: {str(e)}")
            return f"搜索过程中出现错误: {str(e)}"

    def clear_history(self):
        """清空对话历史"""
        self.conversation_history = []
        logger.info("对话历史已清空")

    def get_conversation_history(self) -> List[Dict[str, str]]:
        """获取对话历史"""
        return self.conversation_history

    def get_trace(self) -> List[Dict[str, Any]]:
        """获取上一次 search_and_answer 的 ReAct 轨迹（思考/行动/观察/最终答案）"""
        return self.trace

    def set_temperature(self, temperature: float):
        """
        设置温度参数

        Args:
            temperature: 温度值 (0.0 - 2.0)
        """
        if 0.0 <= temperature <= 2.0:
            self.temperature = temperature
            logger.info(f"温度设置为: {temperature}")
        else:
            logger.warning(f"无效的温度值: {temperature}，应在 0.0 到 2.0 之间")


def run_offline_demo(question: str = "Moonshot AI 的 Context Caching 是什么技术？",
                     verbose: bool = True) -> Dict[str, Any]:
    """离线演示 ReAct 循环——无需 API Key 或联网。

    本函数**不调用真实搜索**，而是回放一段“示例轨迹”，用来直观展示本章讲的
    “想→做→看→想→做→看”循环：模型先思考，再调用 $web_search 行动，观察结果后
    继续思考，最终综合出答案。轨迹内容仅为教学示例，不代表真实搜索返回。

    Returns:
        包含 question / trace / answer 的字典。
    """
    trace: List[Dict[str, Any]] = [
        {"iteration": 1, "type": "thought",
         "content": "用户想了解 Context Caching。这是 Moonshot 的特性，我需要先搜索官方说明，确认它的定义和作用。"},
        {"iteration": 1, "type": "action", "tool": "$web_search",
         "args": {"query": "Moonshot AI Context Caching 是什么"}},
        {"iteration": 1, "type": "observation", "tool": "$web_search",
         "content": "（示例结果）Context Caching 是一种上下文缓存机制：把重复使用的前缀"
                    "（如长系统提示、文档）缓存在服务端，后续请求命中缓存即可复用，"
                    "从而降低重复计算与费用。"},
        {"iteration": 2, "type": "thought",
         "content": "已知大致定义，但还缺少适用场景。再搜一次它的典型用途以便答得更完整。"},
        {"iteration": 2, "type": "action", "tool": "$web_search",
         "args": {"query": "Context Caching 适用场景 计费"}},
        {"iteration": 2, "type": "observation", "tool": "$web_search",
         "content": "（示例结果）常见于多轮对话、长文档反复问答、固定系统提示等场景；"
                    "命中缓存的 token 通常按更低价格计费，并能显著降低首字延迟。"},
        {"iteration": 3, "type": "answer",
         "content": "Context Caching（上下文缓存）是 Moonshot AI 提供的一种机制：将重复使用的"
                    "上下文前缀缓存在服务端，后续请求复用缓存内容，从而降低重复计算、减少费用、"
                    "并加快响应。它特别适合长系统提示、长文档反复问答、多轮对话等场景。"
                    "（本段来自离线示例轨迹，非真实搜索结果。）"},
    ]

    if verbose:
        for step in trace:
            print(format_trace_step(step))

    answer = next(s["content"] for s in trace if s["type"] == "answer")
    return {"question": question, "trace": trace, "answer": answer}


# 独立运行示例
def main():
    """
    独立运行示例，演示基本用法
    """
    # 设置 API key (确保已设置环境变量 MOONSHOT_API_KEY)
    agent = WebSearchAgent()

    # 示例问题
    test_question = "请搜索 Moonshot AI Context Caching 技术，告诉我这是什么。"

    print(f"问题: {test_question}")
    print("-" * 60)
    print("搜索中...")

    # 获取答案
    answer = agent.search_and_answer(test_question)

    print("\n答案:")
    print("-" * 60)
    print(answer)


if __name__ == '__main__':
    main()

config.py

"""
配置文件 - Kimi API 配置
"""

import os
from typing import Optional


def map_model_to_openrouter(model: str) -> str:
    """Map a bare model id to an OpenRouter model id.
    - ids already containing '/' -> left as-is
    - gpt-*/o1-*/o3-*/o4-* -> 'openai/<id>'
    - claude-* -> anthropic Claude (opus/sonnet/haiku)
    - other native ids (kimi-*, doubao-*, ...) are NOT reliably on OpenRouter,
      so fall back to OPENROUTER_MODEL or a safe default that always works.
    """
    m = (model or "").strip()
    if "/" in m:
        return m
    ml = m.lower()
    if ml.startswith(("gpt-", "o1-", "o3-", "o4-")):
        return "openai/" + m
    if ml.startswith("claude-"):
        if "sonnet" in ml:
            return "anthropic/claude-sonnet-4.6"
        if "haiku" in ml:
            return "anthropic/claude-haiku-4.5"
        return "anthropic/claude-opus-4.8"
    if ml.startswith("kimi"):
        # kimi-k3 is not on OpenRouter; moonshotai/kimi-k2.6 is the closest hosted id.
        return "moonshotai/kimi-k2.6"
    return os.getenv("OPENROUTER_MODEL", "openai/gpt-5.6-luna")


def resolve_llm_backend(primary_key: str, primary_base_url: str, model: str):
    """Universal OpenRouter fallback for LLM backend resolution.

    Returns (api_key, base_url, model, using_openrouter).
    - If the primary provider key is present, behavior is unchanged.
    - Else if OPENROUTER_API_KEY is present, route through OpenRouter and map
      the model id to an OpenRouter id.
    - Else raise a clear error listing the accepted keys.
    """
    openrouter_key = os.getenv("OPENROUTER_API_KEY")
    # gpt-5.x (incl. gpt-5.6*) needs OpenAI org-verification on the direct API;
    # when an OpenRouter key is present, prefer routing these ids through it.
    if openrouter_key and str(model or "").lower().startswith("gpt-5"):
        base_url = os.getenv("OPENROUTER_BASE_URL", "https://openrouter.ai/api/v1")
        return openrouter_key, base_url, map_model_to_openrouter(model), True
    if primary_key:
        return primary_key, primary_base_url, model, False
    if openrouter_key:
        base_url = os.getenv("OPENROUTER_BASE_URL", "https://openrouter.ai/api/v1")
        return openrouter_key, base_url, map_model_to_openrouter(model), True
    raise ValueError(
        "No API key found. Set MOONSHOT_API_KEY / KIMI_API_KEY (primary) or "
        "OPENROUTER_API_KEY (universal fallback)."
    )


class Config:
    """配置类"""

    # Kimi API 配置
    MOONSHOT_API_KEY: str = os.getenv("MOONSHOT_API_KEY", "")
    # 向后兼容：如果没有 MOONSHOT_API_KEY，尝试使用 KIMI_API_KEY
    if not MOONSHOT_API_KEY:
        MOONSHOT_API_KEY = os.getenv("KIMI_API_KEY", "")

    KIMI_BASE_URL: str = "https://api.moonshot.cn/v1"

    # 模型配置
    DEFAULT_MODEL: str = "kimi-k3"  # 使用最新的 Kimi K3 模型

    # 搜索配置
    MAX_SEARCH_ITERATIONS: int = 5  # 最大搜索迭代次数（与 agent 默认值保持一致）
    SEARCH_TIMEOUT: int = 30  # 搜索超时时间（秒）

    # 日志配置
    LOG_LEVEL: str = "INFO"
    LOG_FORMAT: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"

    @classmethod
    def validate(cls) -> bool:
        """
        验证配置是否有效

        Returns:
            bool: 配置是否有效
        """
        if not cls.MOONSHOT_API_KEY:
            print("错误: 未设置 MOONSHOT_API_KEY 环境变量")
            print("请设置环境变量: export MOONSHOT_API_KEY='your-api-key'")
            print("(或者使用旧的环境变量名: export KIMI_API_KEY='your-api-key')")
            return False
        return True

    @classmethod
    def get_api_key(cls, api_key: Optional[str] = None) -> str:
        """
        获取 API Key

        Args:
            api_key: 可选的 API key，如果提供则使用，否则从环境变量获取

        Returns:
            API key
        """
        if api_key:
            return api_key
        return cls.MOONSHOT_API_KEY

examples.py

"""
高级示例 - 展示 Web Search Agent 的各种用法
"""

import asyncio
import json
from typing import List, Dict, Any
from agent import WebSearchAgent
from config import Config
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class AdvancedWebSearchAgent(WebSearchAgent):
    """
    高级 Web Search Agent - 扩展功能
    """

    def batch_search(self, questions: List[str]) -> List[Dict[str, str]]:
        """
        批量搜索多个问题

        Args:
            questions: 问题列表

        Returns:
            答案列表
        """
        results = []
        for i, question in enumerate(questions, 1):
            logger.info(f"处理问题 {i}/{len(questions)}: {question}")
            try:
                answer = self.search_and_answer(question)
                results.append({
                    "question": question,
                    "answer": answer,
                    "status": "success"
                })
            except Exception as e:
                results.append({
                    "question": question,
                    "answer": str(e),
                    "status": "error"
                })
            # 清空历史，避免上下文混淆
            self.clear_history()
        return results

    def search_with_context(self, question: str, context: str) -> str:
        """
        带上下文的搜索

        Args:
            question: 用户问题
            context: 额外的上下文信息

        Returns:
            答案
        """
        # 构建带上下文的问题
        contextualized_question = f"""
背景信息：{context}

基于上述背景，请回答以下问题：
{question}
"""
        return self.search_and_answer(contextualized_question)

    def comparative_search(self, items: List[str], aspect: str) -> str:
        """
        比较搜索 - 搜索并比较多个项目

        Args:
            items: 要比较的项目列表
            aspect: 比较的方面

        Returns:
            比较结果
        """
        # 构建比较问题
        items_str = "、".join(items)
        question = f"请搜索并比较 {items_str} 在 {aspect} 方面的差异和优劣"

        return self.search_and_answer(question)

    def fact_check(self, statement: str) -> Dict[str, Any]:
        """
        事实核查 - 验证陈述的真实性

        Args:
            statement: 需要验证的陈述

        Returns:
            验证结果
        """
        question = f"""
请验证以下陈述的真实性：
"{statement}"

请提供：
1. 这个陈述是否准确（真/假/部分真实）
2. 相关的事实和证据
3. 信息来源
"""
        answer = self.search_and_answer(question)

        # 简单解析结果
        is_true = "真" in answer[:100]
        return {
            "statement": statement,
            "is_true": is_true,
            "explanation": answer
        }


def example_basic_search():
    """基础搜索示例"""
    print("\n" + "="*60)
    print("📌 示例 1: 基础搜索")
    print("="*60)

    agent = WebSearchAgent(Config.get_api_key())

    questions = [
        "OpenAI 最新发布的 GPT 模型有什么特点？",
        "如何学习机器学习？推荐一些资源",
    ]

    for q in questions:
        print(f"\n问题: {q}")
        print("-"*40)
        answer = agent.search_and_answer(q)
        print(f"答案: {answer}")


def example_batch_search():
    """批量搜索示例"""
    print("\n" + "="*60)
    print("📌 示例 2: 批量搜索")
    print("="*60)

    agent = AdvancedWebSearchAgent(Config.get_api_key())

    questions = [
        "React 和 Vue 的主要区别是什么？",
        "Python 最适合做什么类型的项目？",
        "如何开始学习人工智能？",
    ]

    results = agent.batch_search(questions)

    for result in results:
        print(f"\n问题: {result['question']}")
        print(f"状态: {result['status']}")
        print(f"答案: {result['answer'][:200]}...")  # 只显示前200字符


def example_contextual_search():
    """带上下文的搜索示例"""
    print("\n" + "="*60)
    print("📌 示例 3: 带上下文的搜索")
    print("="*60)

    agent = AdvancedWebSearchAgent(Config.get_api_key())

    context = "我是一个刚开始学习编程的大学生，主要对 Web 开发感兴趣"
    question = "我应该先学习哪种编程语言？"

    print(f"上下文: {context}")
    print(f"问题: {question}")
    print("-"*40)

    answer = agent.search_with_context(question, context)
    print(f"答案: {answer}")


def example_comparative_search():
    """比较搜索示例"""
    print("\n" + "="*60)
    print("📌 示例 4: 比较搜索")
    print("="*60)

    agent = AdvancedWebSearchAgent(Config.get_api_key())

    # 比较不同的技术框架
    items = ["TensorFlow", "PyTorch", "JAX"]
    aspect = "性能和易用性"

    print(f"比较项目: {', '.join(items)}")
    print(f"比较方面: {aspect}")
    print("-"*40)

    result = agent.comparative_search(items, aspect)
    print(f"比较结果:\n{result}")


def example_fact_check():
    """事实核查示例"""
    print("\n" + "="*60)
    print("📌 示例 5: 事实核查")
    print("="*60)

    agent = AdvancedWebSearchAgent(Config.get_api_key())

    statements = [
        "Python 是世界上最流行的编程语言",
        "量子计算机已经可以破解所有现代加密算法",
        "GPT-4 有 1.76 万亿个参数",
    ]

    for statement in statements:
        print(f"\n陈述: {statement}")
        result = agent.fact_check(statement)
        print(f"真实性: {'✅ 真' if result['is_true'] else '❌ 假/存疑'}")
        print(f"解释: {result['explanation'][:200]}...")


def example_research_assistant():
    """研究助手示例 - 深度研究某个主题"""
    print("\n" + "="*60)
    print("📌 示例 6: 研究助手 - 深度研究")
    print("="*60)

    agent = AdvancedWebSearchAgent(Config.get_api_key())

    topic = "大语言模型的发展历程"

    # 构建研究问题序列
    research_questions = [
        f"什么是{topic}？请提供详细定义",
        f"{topic}的关键里程碑和重要事件有哪些？",
        f"{topic}面临的主要挑战是什么？",
        f"{topic}的未来发展趋势如何？",
    ]

    print(f"研究主题: {topic}")
    print("="*60)

    research_report = []
    for i, q in enumerate(research_questions, 1):
        print(f"\n研究问题 {i}: {q}")
        print("-"*40)
        answer = agent.search_and_answer(q)
        research_report.append({
            "section": i,
            "question": q,
            "findings": answer
        })
        print(f"发现: {answer[:300]}...")
        agent.clear_history()  # 清空历史，确保每个问题独立

    # 保存研究报告
    with open("research_report.json", "w", encoding="utf-8") as f:
        json.dump(research_report, f, ensure_ascii=False, indent=2)
    print(f"\n✅ 研究报告已保存到 research_report.json")


def main():
    """运行所有示例"""

    if not Config.validate():
        print("请先设置 KIMI_API_KEY 环境变量")
        return

    examples = [
        ("基础搜索", example_basic_search),
        ("批量搜索", example_batch_search),
        ("带上下文搜索", example_contextual_search),
        ("比较搜索", example_comparative_search),
        ("事实核查", example_fact_check),
        ("研究助手", example_research_assistant),
    ]

    print("\n" + "="*60)
    print("🎯 Kimi Web Search Agent - 高级示例")
    print("="*60)
    print("\n选择要运行的示例:")

    for i, (name, _) in enumerate(examples, 1):
        print(f"{i}. {name}")
    print(f"{len(examples) + 1}. 运行所有示例")
    print("0. 退出")

    try:
        choice = input("\n请输入选项 (0-7): ").strip()
        choice = int(choice)

        if choice == 0:
            print("退出程序")
            return
        elif 1 <= choice <= len(examples):
            examples[choice - 1][1]()
        elif choice == len(examples) + 1:
            for name, func in examples:
                try:
                    func()
                except Exception as e:
                    logger.error(f"运行 {name} 时出错: {str(e)}")
        else:
            print("无效的选项")
    except ValueError:
        print("请输入有效的数字")
    except KeyboardInterrupt:
        print("\n程序被中断")
    except Exception as e:
        logger.error(f"运行示例时出错: {str(e)}")


if __name__ == "__main__":
    main()

main.py

"""
主程序 - Web Search Agent 使用示例

演示第一章的 ReAct 循环（Reasoning + Acting）：模型先思考，再调用 $web_search
行动，观察搜索结果后继续思考，直到综合出最终答案。运行时会逐步打印 ReAct 轨迹。
"""

import os
import sys
import json
import argparse
import logging
from typing import Optional
from agent import WebSearchAgent, run_offline_demo
from config import Config

# 设置日志
logging.basicConfig(
    level=getattr(logging, Config.LOG_LEVEL),
    format=Config.LOG_FORMAT
)
logger = logging.getLogger(__name__)


def _save_output(path: str, payload: dict):
    """把问题、ReAct 轨迹和答案保存为 JSON 文件"""
    with open(path, "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)
    print(f"\n💾 结果已保存到: {path}")


def run_interactive_mode(agent: WebSearchAgent, output: Optional[str] = None):
    """
    交互式模式 - 持续与 Agent 对话

    Args:
        agent: WebSearchAgent 实例
        output: 可选，保存每次问答轨迹的 JSON 文件路径
    """
    print("\n" + "="*60)
    print("🤖 Kimi Web Search Agent - 交互模式")
    print("="*60)
    print("输入您的问题，Agent 将自动搜索并回答")
    print("输入 'quit' 或 'exit' 退出")
    print("输入 'clear' 清空对话历史")
    print("="*60 + "\n")

    while True:
        try:
            # 获取用户输入
            user_input = input("您的问题: ").strip()

            # 检查退出命令
            if user_input.lower() in ['quit', 'exit', 'q']:
                print("\n👋 再见！")
                break

            # 检查清空命令
            if user_input.lower() == 'clear':
                agent.clear_history()
                print("✅ 对话历史已清空\n")
                continue

            # 检查空输入
            if not user_input:
                print("❌ 请输入一个问题\n")
                continue

            # 显示思考中
            print("\n🔍 Agent 正在搜索和思考（ReAct 轨迹如下）...\n")

            # 获取答案（verbose=True 时轨迹已在 agent 内实时打印）
            answer = agent.search_and_answer(user_input, max_iterations=Config.MAX_SEARCH_ITERATIONS)

            # 显示答案
            print("\n" + "="*60)
            print("📝 Agent 回答:")
            print("-"*60)
            print(answer)
            print("="*60 + "\n")

            if output:
                _save_output(output, {"question": user_input,
                                      "trace": agent.get_trace(),
                                      "answer": answer})

        except KeyboardInterrupt:
            print("\n\n👋 检测到中断，退出程序")
            break
        except Exception as e:
            logger.error(f"处理问题时出错: {str(e)}")
            print(f"\n❌ 出错了: {str(e)}\n")


def run_single_question(agent: WebSearchAgent, question: str,
                        max_iterations: int, output: Optional[str] = None):
    """
    单个问题模式 - 回答一个问题后退出

    Args:
        agent: WebSearchAgent 实例
        question: 要回答的问题
        max_iterations: 最大 ReAct 迭代次数
        output: 可选，保存轨迹的 JSON 文件路径
    """
    print("\n" + "="*60)
    print("🤖 Kimi Web Search Agent")
    print("="*60)
    print(f"问题: {question}")
    print("-"*60)
    print("🔍 ReAct 轨迹（想 → 做 → 看）:\n")

    try:
        answer = agent.search_and_answer(question, max_iterations=max_iterations)
        print("\n📝 答案:")
        print("-"*60)
        print(answer)
        print("="*60 + "\n")

        if output:
            _save_output(output, {"question": question,
                                  "trace": agent.get_trace(),
                                  "answer": answer})
    except Exception as e:
        logger.error(f"处理问题时出错: {str(e)}")
        print(f"\n❌ 出错了: {str(e)}\n")


def build_parser() -> argparse.ArgumentParser:
    """构建命令行参数解析器（中文帮助）"""
    parser = argparse.ArgumentParser(
        prog="main.py",
        description="Kimi Web Search Agent —— 演示 ReAct 循环（思考→行动→观察）的搜索 Agent。",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""示例:
  python main.py                                  # 进入交互模式
  python main.py "2024 诺贝尔物理学奖得主是谁？"    # 单次问答，打印 ReAct 轨迹
  python main.py --provider offline-demo          # 离线演示 ReAct 循环（无需 API Key）
  python main.py "比特币现价" --max-steps 3 --output result.json
""",
    )
    parser.add_argument("query", nargs="*",
                        help="要提问的问题；省略则进入交互模式")
    parser.add_argument("--provider", choices=["kimi", "offline-demo"], default="kimi",
                        help="搜索后端：kimi=调用 Kimi 内置 $web_search（需 API Key）；"
                             "offline-demo=离线回放示例轨迹（默认 kimi）")
    parser.add_argument("--model", default=Config.DEFAULT_MODEL,
                        help=f"使用的模型名称（默认 {Config.DEFAULT_MODEL}）")
    parser.add_argument("--max-steps", type=int, default=Config.MAX_SEARCH_ITERATIONS,
                        help=f"最大 ReAct 迭代次数（默认 {Config.MAX_SEARCH_ITERATIONS}）")
    parser.add_argument("--base-url", default=Config.KIMI_BASE_URL,
                        help=f"API 基础 URL（默认 {Config.KIMI_BASE_URL}）")
    parser.add_argument("--api-key", default=None,
                        help="Kimi API Key（默认从 MOONSHOT_API_KEY / KIMI_API_KEY 环境变量读取）")
    parser.add_argument("--output", "-o", default=None,
                        help="将问题、ReAct 轨迹和答案保存到指定 JSON 文件")
    parser.add_argument("--quiet", action="store_true",
                        help="不实时打印 ReAct 轨迹（默认打印）")
    return parser


def main(argv: Optional[list] = None):
    """主函数：解析命令行参数并分发到相应模式"""
    parser = build_parser()
    args = parser.parse_args(argv)
    question = " ".join(args.query).strip()

    # 离线演示模式：无需 API Key，回放示例轨迹展示 ReAct 循环
    if args.provider == "offline-demo":
        demo_question = question or "Moonshot AI 的 Context Caching 是什么技术？"
        print("\n" + "="*60)
        print("🧪 离线演示模式（示例轨迹，非真实搜索结果）")
        print("="*60)
        print(f"问题: {demo_question}")
        print("-"*60)
        print("🔍 ReAct 轨迹（想 → 做 → 看）:\n")
        result = run_offline_demo(demo_question, verbose=not args.quiet)
        print("\n📝 答案:")
        print("-"*60)
        print(result["answer"])
        print("="*60 + "\n")
        if args.output:
            _save_output(args.output, result)
        return

    # 在线模式：需要 API Key
    api_key = Config.get_api_key(args.api_key)
    if not api_key and not os.getenv("OPENROUTER_API_KEY"):
        Config.validate()
        print("提示：也可设置 OPENROUTER_API_KEY 作为通用兜底。")
        sys.exit(1)

    # 创建 Agent
    try:
        agent = WebSearchAgent(
            api_key=api_key,
            base_url=args.base_url,
            model=args.model,
            verbose=not args.quiet,
        )
        logger.info("Agent 初始化成功")
    except Exception as e:
        logger.error(f"Agent 初始化失败: {str(e)}")
        sys.exit(1)

    # 有问题则单次问答，否则进入交互模式
    if question:
        run_single_question(agent, question, args.max_steps, args.output)
    else:
        run_interactive_mode(agent, args.output)


if __name__ == "__main__":
    main()

quickstart.py

#!/usr/bin/env python3
"""
快速开始脚本 - 一键体验 Kimi Web Search Agent
"""

import os
import sys
from agent import WebSearchAgent
from config import Config

# 彩色输出
class Colors:
    HEADER = '\033[95m'
    BLUE = '\033[94m'
    CYAN = '\033[96m'
    GREEN = '\033[92m'
    WARNING = '\033[93m'
    FAIL = '\033[91m'
    END = '\033[0m'
    BOLD = '\033[1m'


def print_colored(text, color):
    """打印彩色文本"""
    print(f"{color}{text}{Colors.END}")


def print_banner():
    """打印欢迎横幅"""
    banner = """
╔══════════════════════════════════════════════════════════╗
║         🤖 Kimi Web Search Agent - 快速体验              ║
║                                                          ║
║  基于 Kimi API 的智能搜索助手                             ║
║  能够自动搜索网络信息并生成智能答案                         ║
╚══════════════════════════════════════════════════════════╝
"""
    print_colored(banner, Colors.CYAN)


def check_api_key():
    """检查 API Key 配置"""
    api_key = os.getenv("MOONSHOT_API_KEY")
    if not api_key:
        # 向后兼容：尝试旧的环境变量名
        api_key = os.getenv("KIMI_API_KEY")

    if not api_key:
        print_colored("\n⚠️  未检测到 API Key", Colors.WARNING)
        print("\n请按以下步骤配置:")
        print("1. 访问 https://platform.moonshot.ai/ 获取 API Key")
        print("2. 设置环境变量:")
        print("   export MOONSHOT_API_KEY='your-api-key'")
        print("   (或使用: export KIMI_API_KEY='your-api-key')")
        print("\n或者直接输入 API Key (输入 'skip' 跳过):")

        user_input = input("> ").strip()

        if user_input.lower() == 'skip':
            return None
        elif user_input:
            return user_input
        else:
            return None

    print_colored("✅ API Key 已配置", Colors.GREEN)
    return api_key


def demo_search(agent):
    """演示搜索功能"""
    print_colored("\n📝 演示搜索功能", Colors.HEADER)
    print("-" * 60)

    demo_questions = [
        "OpenAI 最新发布了什么产品？",
        "2024年有哪些重要的AI突破？",
        "如何开始学习机器学习？",
    ]

    print("选择一个演示问题，或输入您自己的问题:")
    for i, q in enumerate(demo_questions, 1):
        print(f"{i}. {q}")
    print("0. 输入自定义问题")

    choice = input("\n请选择 (0-3): ").strip()

    try:
        choice = int(choice)
        if choice == 0:
            question = input("请输入您的问题: ").strip()
            if not question:
                print_colored("❌ 问题不能为空", Colors.FAIL)
                return
        elif 1 <= choice <= len(demo_questions):
            question = demo_questions[choice - 1]
        else:
            print_colored("❌ 无效的选择", Colors.FAIL)
            return
    except ValueError:
        print_colored("❌ 请输入数字", Colors.FAIL)
        return

    print_colored(f"\n🔍 正在搜索: {question}", Colors.BLUE)
    print("请稍候，Agent 正在搜索和分析...")
    print("-" * 60)

    try:
        answer = agent.search_and_answer(question)
        print_colored("\n📖 Agent 回答:", Colors.GREEN)
        print(answer)
    except Exception as e:
        print_colored(f"\n❌ 搜索失败: {str(e)}", Colors.FAIL)


def interactive_mode(agent):
    """交互模式"""
    print_colored("\n💬 进入交互模式", Colors.HEADER)
    print("您可以连续提问，输入 'quit' 退出")
    print("-" * 60)

    while True:
        question = input("\n您的问题: ").strip()

        if question.lower() in ['quit', 'exit', 'q']:
            print_colored("👋 感谢使用！", Colors.GREEN)
            break

        if not question:
            continue

        print_colored("🔍 搜索中...", Colors.BLUE)

        try:
            answer = agent.search_and_answer(question)
            print_colored("\n📖 回答:", Colors.GREEN)
            print(answer)
        except Exception as e:
            print_colored(f"❌ 错误: {str(e)}", Colors.FAIL)


def main():
    """主函数"""
    print_banner()

    # 检查 API Key
    api_key = check_api_key()
    if not api_key:
        print_colored("\n⚠️  无法继续，需要配置 API Key", Colors.WARNING)
        sys.exit(1)

    # 创建 Agent
    try:
        print_colored("\n🚀 初始化 Agent...", Colors.BLUE)
        agent = WebSearchAgent(api_key=api_key)
        print_colored("✅ Agent 已就绪", Colors.GREEN)
    except Exception as e:
        print_colored(f"❌ 初始化失败: {str(e)}", Colors.FAIL)
        sys.exit(1)

    # 选择模式
    print("\n选择使用模式:")
    print("1. 演示搜索 (快速体验)")
    print("2. 交互模式 (连续对话)")
    print("3. 退出")

    mode = input("\n请选择 (1-3): ").strip()

    if mode == "1":
        demo_search(agent)
        # 询问是否继续
        cont = input("\n是否进入交互模式？(y/n): ").strip().lower()
        if cont == 'y':
            interactive_mode(agent)
    elif mode == "2":
        interactive_mode(agent)
    elif mode == "3":
        print_colored("👋 再见！", Colors.GREEN)
    else:
        print_colored("❌ 无效的选择", Colors.FAIL)

    print_colored("\n感谢使用 Kimi Web Search Agent！", Colors.CYAN)
    print("更多功能请查看:")
    print("- README.md: 完整文档")
    print("- examples.py: 高级示例")
    print("- main.py: 主程序")


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print_colored("\n\n👋 程序被中断", Colors.WARNING)
    except Exception as e:
        print_colored(f"\n❌ 发生错误: {str(e)}", Colors.FAIL)
        sys.exit(1)