AI News
中文资讯 掘金 AI Sat, 25 Jul 2026 正文 ok 1257 words

DeepResearchSystem 0x01：Agent 基础实现

英文标题：DeepResearchSystem 0x01：Agent 基础

本文延续 DeepResearchSystem 的项目背景与整体设计，开始实现 Agent 的基础能力。内容涵盖通过 `.env` 管理 LLM 与 MCP 配置、加载可用模型及默认模型的兜底逻辑。文章实现了基础 Agent、JsonAgent、基于令牌桶算法的 API 限流器，以及 WebSearch MCP 调用与简单重试机制。对开发者而言，这些组件构成了研究型 Agent 系统中模型调用、结构化输出、请求限流和多厂商 MCP 响应兼容解析的基础设施。
打开原文相关 3 · 机会 1
media3.2 KB media589 B media11.7 KB media63.6 KB media65.2 KB

中文

前言

上一篇 主要介绍了项目背景和整体构思。这一章废话少说，直接开始动手实现。

.env

和以往一样，项目涉及的 LLM api_key 等信息不会硬编码到代码中，而是放在环境文件中。

LLM_APP_KEY = "你的LLM API Key" LLM_BASE_URL = "你的LLM服务URL"

MCP_APP_ID = "你的MCP工具ID"

AVAILABLE_MODELS=[{"model_id":"qwen3.6-flash","display_name":"Qwen-Flash","icon":"Zap","icon_color":"yellow-400"},{"model_id":"qwen3.6-plus","display_name":"Qwen-Plus","icon":"Zap","icon_color":"orange-400"},{"model_id":"qwen3.7-max","display_name":"Qwen-Max","icon":"Cpu","icon_color":"purple-400"}]

WEB_SEARCH_MAX_QPS = 12

NUMBER_OF_INITIAL_QUERIES = 2

MAX_RESEARCH_LOOPS = 2

LLM

该模块主要封装客户端与 LLM 的通信。其实很简单，在过去的各种 demo 中已经非常熟悉这类实现了。

ModelConfig

定义模型相关配置，用于在 LLM 调用中唯一确认使用哪个厂商的哪个 LLM。

class ModelConfig(BaseModel) : """LLM模型配置""" id : str = Field(..., description="模型ID") name : str = Field(..., description="模型名称") icon :str = Field(default="Zap", description="图标类型(Zap/Cpu)") icon_color : str = Field(default="yellow-400", description="图标颜色")

加载可用模型

前面提到，涉及 LLM 的 api_key 存储在环境文件中。这里需要从环境变量加载可用模型。一定要注意兜底模型的配置：当环境变量未配置时，代码不能因此被阻塞。

def load_available_models_from_env() : """从环境变量加载可用模型""" model_json = os.getenv(AVAILABLE_MODELS)

if not model_json :
    
    return [
        ModelConfig(id="qwen3.6-flash", name="Qwen-Flash", icon="Zap", icon_color="yellow-400"),
        ModelConfig(id="qwen3.6-plus", name="Qwen-Plus", icon="Zap", icon_color="green-400"),
        ModelConfig(id="qwen3.7-max", name="Qwen-Max", icon="Cpu", icon_color="blue-400")
    ]

try:
    models_data = json.loads(model_json)
    return [ModelConfig(**model) for model in models_data]
except Exception as e :
    print(f"警告: 解析AVAILABLE_MODELS失败，使用默认模型列表。错误: {e}")
    return [
        ModelConfig(id="qwen3.6-flash", name="Qwen-Flash", icon="Zap", icon_color="yellow-400"),
        ModelConfig(id="qwen3.6-plus", name="Qwen-Plus", icon="Zap", icon_color="green-400"),
        ModelConfig(id="qwen3.7-max", name="Qwen-Max", icon="Cpu", icon_color="blue-400")
    ]

默认模型

def get_default_model_id() : """获取默认模型ID（模型列表的最后一项）""" models = load_available_models_from_env() if models : return models[-1].id return "qwen3.7-max"

Agent

Agent 负责承接上层具体业务，封装 LLM 并将提示词传给 LLM，同时也负责处理 LLM 返回的数据。其核心 LLM 调用在于 step 方法，该方法接收提示词和其他关键信息。

class Agent : step_prompt = """{prompt}""" def init(self, model_id=get_default_model_id()) : self.llm = OpenAICompatibleLLM(model_id=model_id)

def __call(self, prompt) :
    response = self.llm.generate_response(prompt)
    return response

def set_step_prompt(self, prompt):
    self.step_prompt = prompt

def step(self, **kwargs):
    step_prompt = self.prompt_format(self.step_prompt, **kwargs)
    response = ""
    for _ in range(3) :
        try:
            response = self(step_prompt)
            response = self.post_process(response)
            break
        except Exception as e :
            logger.error(f"大模型调用错误：{e}\n{traceback.format_exc()}")
            continue

        return response

def post_process(self, response):
    return response

Prompt 模板渲染

注意，这里主要完成将 kwargs 中的关键参数高效替换到 Prompt 的 {xxx} 占位符中。

def prompt_format(self, prompt, **kwargs) : """高效进行 Prompt 模板渲染的基础方式"""

prompt_ = copy.deepcopy(prompt)

for k in  kwargs.keys() :
    
    rep = "{" + k + "}"
    prompt_ = prompt_.replace(rep, str(kwargs[k]))

return prompt_

JsonAgent

Agent 的子类，主要负责将 LLM 返回数据转换为 JSON 格式，核心代码位于 post_process 方法。

class JsonAgent(Agent): """将LLM返回数据转换为JSON根式""" def init(self, model_id=get_default_model_id(), keys=None): super().init(model_id) self.keys = keys

def post_process(self, response):
    """self.keys参数转换为Pydantic模型类"""
    result = json.loads(jsonUtils.extract_pattern(response, pattern="json"))
    if not self.keys:
        return result
    
    
    return self.keys(**result)

限流器

什么是限流器？主流 LLM 对外部请求通常都会设置限流。例如，1 秒内连续发起 N 次请求可能会触发 429 错误。因此，在请求 LLM 时，需要通过限流器控制自身发起的 LLM 请求，以减少触发 429 错误的概率。目前实现的 RateLimiter 基于令牌桶算法。

令牌桶算法是网络流量整形（Traffic Shaping）和速率限制（Rate Limiting）中最经典的算法之一。

  • 1. 有一个桶：系统以恒定速率，例如 max_qps=15，向桶中放入“令牌”。

  • 2. 令牌满了就溢出：桶具有最大容量（Burst Size）。如果桶已满，新生成的令牌会被丢弃。

  • 3. 请求拿令牌：每当一个请求到来，必须从桶中取走一个令牌。

  • 4. 没有令牌就等待/拒绝：如果桶为空，请求必须等待，直到生成新令牌，或者直接被拒绝。

  • 本质：这是一个允许短期突发的平均速率限制器。长期速率趋近于 rate = 令牌生成速率，但瞬时可以处理 burst = 桶容量 个请求。

一个例子：门一直开着（非阻塞），但有一名保安（令牌桶算法）负责发放准入手环（令牌）。手环以固定速率生成，例如每 66ms 生成一个，并放在一个盒子里（桶）。

class TokenBucketRateLimiter: """ 基于标准令牌桶算法的速率限制器 支持突发流量，用于控制API请求频率，避免触发429错误 """

def __init__(self, max_qps: float = 15.0, burst_capacity: float | None = None):
    """
    初始化令牌桶速率限制器
    :param max_qps: 令牌生成速率（每秒请求数）
    :param burst_capacity: 桶容量（最大突发请求数），默认等于max_qps，即允许1秒内的全部突发
    """
    self.max_qps = max_qps
    self.burst_capacity = burst_capacity if burst_capacity is not None else max_qps
    
    self.current_tokens = float(self.burst_capacity)
    
    self.last_refill_time = time.monotonic()
    self.lock = asyncio.Lock()
    logger.info(
        f"令牌桶速率限制器初始化：QPS={max_qps}, "
        f"桶容量={self.burst_capacity}, "
        f"单令牌生成间隔={1.0 / max_qps:.3f}秒"
    )

async def acquire(self) -> float:
    """
    阻塞式获取请求许可，若令牌不足则等待至有足够令牌
    :return: float 实际等待时间（秒）
    """
    wait_time = 0.0
    async with self.lock:
        now = time.monotonic()
        
        
        elapsed = now - self.last_refill_time
        
        new_tokens = elapsed * self.max_qps
        
        self.current_tokens = min(self.burst_capacity, self.current_tokens + new_tokens)
        
        self.last_refill_time = now

        
        if self.current_tokens >= 1.0:
            
            self.current_tokens -= 1.0
        else:
            
            
            deficit = 1.0 - self.current_tokens
            
            wait_time = deficit / self.max_qps
            
            self.last_refill_time = now + wait_time
            self.current_tokens = 0.0  

    
    if wait_time > 0:
        logger.debug(f"速率限制，需要等待{wait_time:.3f}秒")
        await asyncio.sleep(wait_time)  
    return wait_time

def try_acquire(self) -> bool:
    """
    非阻塞式获取请求许可，若令牌不足立即返回False
    :return: bool 是否成功获取许可
    """
    with self.lock:
        now = time.monotonic()
        
        elapsed = now - self.last_refill_time
        new_tokens = elapsed * self.max_qps
        self.current_tokens = min(self.burst_capacity, self.current_tokens + new_tokens)
        self.last_refill_time = now

        if self.current_tokens >= 1.0:
            self.current_tokens -= 1.0
            return True
        return False

def get_remaining_tokens(self) -> float:
    """返回当前剩余令牌数，仅用于监控调试"""
    with self.lock:
        return self.current_tokens

重试机制

由于存在限流，也就需要重试；但不能立刻进行盲目重试。针对不同错误，应采取不同的重试策略。一言以蔽之：「错误分类 → 退避决策 → 自适应调整 → 熔断兜底」。

上面提到了**「指数退避 + 随机抖动」**，具体指什么？

  • 指数退避：一种逐步拉长重试间隔的策略，公式为：delay = base_delay * (2 ^ attempt)

    • base_delay：基础等待时间（LLM 场景建议从 5s 起步，远高于普通 API 的 1s，因为 LLM 服务商恢复周期更长）。
    • attempt：重试次数（第 1 次重试 attempt=0，第 2 次为 1，以此类推）。
    • 效果：第 1 次等待 5s，第 2 次等待 10s，第 3 次等待 20s，避免在短时间内高频重试并冲击服务端。

  • 随机抖动：用于防止指数退避引发惊群效应。在计算结果上叠加随机因子，打散重试时间，公式为：delay = base_delay * (2 ^ attempt) * random.uniform(1-jitter, 1+jitter)

    • 惊群效应：100 个客户端同时触发 429，都会在第 5s、10s、20s 的整点重试，再次集中打爆服务端，进而陷入「限流 → 重试 → 更严格限流」的死循环。
    • jitter：抖动系数（LLM 场景建议为 0.3~0.5，即 ±30%~50% 的波动范围）。
    • 效果：原本统一的 5s 等待会变为 3.5s~6.5s 之间的随机值，从而错开所有客户端的重试时间。

这里的代码并没有实现复杂的重试机制，只使用了简单的线性退避。

WebSearchAgent

MCPAgent 的子类，主要实现与 WebSearch MCP 的通信。

class WebSearchAgent(MCPAgent): """WebSearch MCP""" async def astep(self, prompt, **kwargs): try: step_prompt = self.step_prompt.format(prompt=prompt) except Exception as e: step_prompt = self.step_prompt

    api_key = os.getenv("LLM_API_KEY")
    app_id = os.getenv("MCP_APP_ID")

    
    rate_limiter = get_web_search_rate_limiter()
    for attempt in range(3):
        try:
            
            wait_time = await rate_limiter.acquire()
            if wait_time > 0 :
                logger.debug(f"速率限制等待：{wait_time:.3f}秒")

            
            
            response = await asyncio.to_thread(
                Application.call,
                api_key=api_key,
                app_id=app_id,
                prompt=step_prompt,
                biz_params=kwargs,
                )
            response = self.extract_pages_from_mcp_response(response, None)
            return response

        except Exception as e:
            error_msg = str(e)
            logger.error(f"Web搜错错误（尝试{attempt + 1} / 3）：{e}\n{traceback.format_exc()}")

            
            
            if "429" in error_msg :
                
                wait_time = 5 * (attempt + 1)
                logger.warning(f"检测到 429 错误，等待 {wait_time} 秒后重试...")
                time.sleep(wait_time)
            
            elif attempt < 2:
                time.sleep(2)
            continue
    return None

多路探测防御性解析

如果频繁使用 MCP，可能会发现不同厂商返回的数据结构存在差异。这导致数据解析逻辑无法一成不变。为解决工具调用返回结构的“不确定性”问题，这里编写了一个函数，对返回数据进行多路探测防御性解析。

  • 使用预定义的 JSONPath 列表递归解析。
  • 使用抽象、归一化的解析函数替代分散的 if/else 代码，统一处理多种结构。
  • 扩展性更好，后续新增路径时，只需更新预定义的 JSONPath 列表。

def extract_pages_from_mcp_response(self, response, config=None): """从MCP WebSearch响应中提取pages数据（重构版） Args: response: MCP响应对象 config: 解析路径配置，默认使用预定义的JSONPath列表 """ if response is None: raise Exception("Web搜索结果不正确") if not response.status_code == 200: raise Exception(f"Web搜索异常: {response.status_code}")

DEFAULT_PARSE_CONFIG = {
    "paths": [
        "$.pages",                  
        "$.data.pages",             
        "$.result.content[0].text",  
        "$.choices[0].message.content",
        
    ]
}

parse_config = config if config else DEFAULT_PARSE_CONFIG

try:
    
    first_level = json.loads(response.output.text)
except json.JSONDecodeError as e:
    logger.error(f"JSON解析失败: {e}, 原始响应: {response.output.text[:500]}")
    raise Exception(f"Web搜索结果JSON解析失败: {str(e)}")

pages = None
matched_path = None


for path_str in parse_config["paths"]:
    try:
        
        jsonpath_expr = jsonpath_parser.parse(path_str)
        matches = [match.value for match in jsonpath_expr.find(first_level)]
        if matches:
            
            candidate = matches[0]

            
            if isinstance(candidate, str):
                try:
                    candidate = json.loads(candidate)
                    
                    
                    if isinstance(candidate, dict) and "pages" in candidate:
                        pages = candidate["pages"]
                        matched_path = path_str + " -> inner.pages"
                        break
                    
                    elif isinstance(candidate, list):
                        pages = candidate
                        matched_path = path_str + " -> inner.list"
                        break
                except json.JSONDecodeError:
                    
                    continue

            
            elif isinstance(candidate, list):
                pages = candidate
                matched_path = path_str
                break

    except JSONPathError as e:
        logger.warning(f"JSONPath语法错误或匹配失败: {path_str}, Error: {e}")
        continue
    except (KeyError, IndexError, TypeError) as e:
        
        continue


if pages is None:
    logger.error(
        f"无法从响应中提取pages数据。尝试的路径: {parse_config['paths']}。"
        f"响应结构: {json.dumps(first_level, ensure_ascii=False)[:500]}"
    )
    raise Exception("无法从Web搜索结果中提取页面数据")

if not isinstance(pages, lis

English / 原文

前言

上一篇 主要讲了下项目背景和整体构思。这一章废话少说，撸起袖子就是干。

.env

和以往一样,我们项目涉及的 LLM 相关 api_key 等信息都不会硬编码到代码中。而是放在环境文件中。

LLM_APP_KEY = "你的LLM API Key"
LLM_BASE_URL = "你的LLM服务URL"


MCP_APP_ID = "你的MCP工具ID"






AVAILABLE_MODELS=[{"model_id":"qwen3.6-flash","display_name":"Qwen-Flash","icon":"Zap","icon_color":"yellow-400"},{"model_id":"qwen3.6-plus","display_name":"Qwen-Plus","icon":"Zap","icon_color":"orange-400"},{"model_id":"qwen3.7-max","display_name":"Qwen-Max","icon":"Cpu","icon_color":"purple-400"}]


WEB_SEARCH_MAX_QPS = 12


NUMBER_OF_INITIAL_QUERIES = 2

MAX_RESEARCH_LOOPS = 2

LLM

该模块主要封装客户端与 LLM 的通讯。其实很简单，我们在过往各种 demo 中已经在熟悉不过了。

ModelConfig

定义模型相关配置，在 LLM 调用中唯一确认使用的是哪家的哪个 LLM。

class ModelConfig(BaseModel) :
    """LLM模型配置"""
    id : str = Field(..., description="模型ID")
    name : str = Field(..., description="模型名称")
    icon :str = Field(default="Zap", description="图标类型(Zap/Cpu)")
    icon_color : str = Field(default="yellow-400", description="图标颜色")

加载可用模型

前面说了涉及 LLM 的 api_key存储在环境文件中。这里就需要从环境变量加载可用模型。这里一定要注意兜底模型 的配置，的那个环境变量未配置时，代码不能因此而 block。

def load_available_models_from_env() :
    """从环境变量加载可用模型"""
    model_json = os.getenv(AVAILABLE_MODELS)

    if not model_json :
        
        return [
            ModelConfig(id="qwen3.6-flash", name="Qwen-Flash", icon="Zap", icon_color="yellow-400"),
            ModelConfig(id="qwen3.6-plus", name="Qwen-Plus", icon="Zap", icon_color="green-400"),
            ModelConfig(id="qwen3.7-max", name="Qwen-Max", icon="Cpu", icon_color="blue-400")
        ]

    try:
        models_data = json.loads(model_json)
        return [ModelConfig(**model) for model in models_data]
    except Exception as e :
        print(f"警告: 解析AVAILABLE_MODELS失败，使用默认模型列表。错误: {e}")
        return [
            ModelConfig(id="qwen3.6-flash", name="Qwen-Flash", icon="Zap", icon_color="yellow-400"),
            ModelConfig(id="qwen3.6-plus", name="Qwen-Plus", icon="Zap", icon_color="green-400"),
            ModelConfig(id="qwen3.7-max", name="Qwen-Max", icon="Cpu", icon_color="blue-400")
        ]

默认模型

def get_default_model_id() :
    """获取默认模型ID（模型列表的最后一项）"""
    models = load_available_models_from_env()
    if models :
        return models[-1].id
    return "qwen3.7-max"

Agent

Agent 负责承接上层具体的业务，封装 LLM 并将提示词传给 LLM。同时也负责对 LLM 返回的数据进行处理。对 LLM 核心调用在于 step方法传入提示词和其他关键信息。

class Agent :
    step_prompt = """{prompt}"""
    def __init__(self, model_id=get_default_model_id()) :
        self.llm = OpenAICompatibleLLM(model_id=model_id)

    def __call(self, prompt) :
        response = self.llm.generate_response(prompt)
        return response

    def set_step_prompt(self, prompt):
        self.step_prompt = prompt

    def step(self, **kwargs):
        step_prompt = self.prompt_format(self.step_prompt, **kwargs)
        response = ""
        for _ in range(3) :
            try:
                response = self(step_prompt)
                response = self.post_process(response)
                break
            except Exception as e :
                logger.error(f"大模型调用错误：{e}\n{traceback.format_exc()}")
                continue

            return response

    def post_process(self, response):
        return response

Prompt 模板渲染

注意，这里主要完成 kwargs关键参数到 prompt {xxx}占位符的高效替换。

def prompt_format(self, prompt, **kwargs) :
    """高效进行 Prompt 模板渲染的基础方式"""
    
    prompt_ = copy.deepcopy(prompt)
    
    for k in  kwargs.keys() :
        
        rep = "{" + k + "}"
        prompt_ = prompt_.replace(rep, str(kwargs[k]))

    return prompt_

JsonAgent

Agent 子类。主要负责将 LLM 返回数据转换为 JSON 根式。核心代码在post_process方法。

class JsonAgent(Agent):
    """将LLM返回数据转换为JSON根式"""
    def __init__(self, model_id=get_default_model_id(), keys=None):
        super().__init__(model_id)
        self.keys = keys

    def post_process(self, response):
        """self.keys参数转换为Pydantic模型类"""
        result = json.loads(jsonUtils.extract_pattern(response, pattern="json"))
        if not self.keys:
            return result
        
        
        return self.keys(**result)

限流器

什么是限流器？ 主流的 LLM 对外部请求都会做限流限制。比如 1s 内连续 N 次请求会触发 429 错误。所以我们在请求 LLM 时，要做一个限流器用于控制自己发起的 LLM 请求，以减少触发 429 错误的概率。我们目前实现的 RateLimiter 基于令牌桶算法。

令牌桶算法是网络流量整形（Traffic Shaping）和速率限制（Rate Limiting）中最经典的算法之一。

  • 1. 有一个桶 ：系统以恒定的速率（如你的 max_qps=15）向桶里放入“令牌”。

  • 2. 令牌满了就溢出 ：桶有最大容量（Burst Size）。如果桶满了，新来的令牌会被丢弃。

  • 3. 请求拿令牌 ：每当一个请求进来，必须从桶里拿走一个令牌。

  • 4. 没令牌就等待/拒绝 ：如果桶空了，请求就必须等待，直到有新的令牌生成，或者直接被拒绝。

  • 本质：允许短期突发的平均速率限制器。长期速率趋近于 rate = 令牌生成速率，但瞬间可以处理 burst = 桶容量 个请求。

一个🌰：门一直开着（非阻塞），但是有个保安（令牌桶算法）在发准入手环（令牌）。手环以固定速率（每66ms一个）生成，放在一个盒子里（桶）。

class TokenBucketRateLimiter:
    """
    基于标准令牌桶算法的速率限制器
    支持突发流量，用于控制API请求频率，避免触发429错误
    """

    def __init__(self, max_qps: float = 15.0, burst_capacity: float | None = None):
        """
        初始化令牌桶速率限制器
        :param max_qps: 令牌生成速率（每秒请求数）
        :param burst_capacity: 桶容量（最大突发请求数），默认等于max_qps，即允许1秒内的全部突发
        """
        self.max_qps = max_qps
        self.burst_capacity = burst_capacity if burst_capacity is not None else max_qps
        
        self.current_tokens = float(self.burst_capacity)
        
        self.last_refill_time = time.monotonic()
        self.lock = asyncio.Lock()
        logger.info(
            f"令牌桶速率限制器初始化：QPS={max_qps}, "
            f"桶容量={self.burst_capacity}, "
            f"单令牌生成间隔={1.0 / max_qps:.3f}秒"
        )

    async def acquire(self) -> float:
        """
        阻塞式获取请求许可，若令牌不足则等待至有足够令牌
        :return: float 实际等待时间（秒）
        """
        wait_time = 0.0
        async with self.lock:
            now = time.monotonic()
            
            
            elapsed = now - self.last_refill_time
            
            new_tokens = elapsed * self.max_qps
            
            self.current_tokens = min(self.burst_capacity, self.current_tokens + new_tokens)
            
            self.last_refill_time = now

            
            if self.current_tokens >= 1.0:
                
                self.current_tokens -= 1.0
            else:
                
                
                deficit = 1.0 - self.current_tokens
                
                wait_time = deficit / self.max_qps
                
                self.last_refill_time = now + wait_time
                self.current_tokens = 0.0  

        
        if wait_time > 0:
            logger.debug(f"速率限制，需要等待{wait_time:.3f}秒")
            await asyncio.sleep(wait_time)  
        return wait_time

    def try_acquire(self) -> bool:
        """
        非阻塞式获取请求许可，若令牌不足立即返回False
        :return: bool 是否成功获取许可
        """
        with self.lock:
            now = time.monotonic()
            
            elapsed = now - self.last_refill_time
            new_tokens = elapsed * self.max_qps
            self.current_tokens = min(self.burst_capacity, self.current_tokens + new_tokens)
            self.last_refill_time = now

            if self.current_tokens >= 1.0:
                self.current_tokens -= 1.0
                return True
            return False

    def get_remaining_tokens(self) -> float:
        """返回当前剩余令牌数，仅用于监控调试"""
        with self.lock:
            return self.current_tokens

重试机制

因为有限流就会有重试，但是不能立刻盲目地重试。针对不同的错误要采取不同的重试策略。一言以蔽之就是：「错误分类→退避决策→自适应调整→熔断兜底」

上面提到了 「指数退避+随机抖动」 ，那么具体是指什么呢？

  • 指数退避：一种逐步拉长重试间隔的策略，公式为：delay = base_delay * (2 ^ attempt)
    • base_delay：基础等待时间（LLM场景推荐5s起步，远大于普通API的1s，因为LLM服务商恢复周期更长）
    • attempt：重试次数（第1次重试attempt=0，第2次=1，以此类推）
    • 效果：第1次等5s，第2次等10s，第3次等20s，避免短时间内高频重试冲击服务端。
  • 随机抖动：防止指数退避引起惊群效应 。在计算结果上叠加随机因子，把重试时间打散，公式为：delay = base_delay * (2 ^ attempt) * random.uniform(1-jitter, 1+jitter)
    • 惊群效应：100个客户端同时触发429，都会在第5s、10s、20s整点重试，再次集中打爆服务端，陷入「限流→重试→更严厉限流」的死循环。
    • jitter：抖动系数（LLM场景推荐0.3~0.5，即±30%~50%的波动范围）
    • 效果：原本统一的5s等待，会变成3.5s~6.5s之间的随机值，错开所有客户端的重试时间。

我们这里的代码并没有实现复杂的重试机制，只是简单的线性退避。

WebSearchAgent

MCPAgent 子类。主要实现和 WebSearch MCP 的通讯。

class WebSearchAgent(MCPAgent):
    """WebSearch MCP"""
    async def astep(self, prompt, **kwargs):
        try:
            step_prompt = self.step_prompt.format(prompt=prompt)
        except Exception as e:
            step_prompt = self.step_prompt

        api_key = os.getenv("LLM_API_KEY")
        app_id = os.getenv("MCP_APP_ID")

        
        rate_limiter = get_web_search_rate_limiter()
        for attempt in range(3):
            try:
                
                wait_time = await rate_limiter.acquire()
                if wait_time > 0 :
                    logger.debug(f"速率限制等待：{wait_time:.3f}秒")

                
                
                response = await asyncio.to_thread(
                    Application.call,
                    api_key=api_key,
                    app_id=app_id,
                    prompt=step_prompt,
                    biz_params=kwargs,
                    )
                response = self.extract_pages_from_mcp_response(response, None)
                return response

            except Exception as e:
                error_msg = str(e)
                logger.error(f"Web搜错错误（尝试{attempt + 1} / 3）：{e}\n{traceback.format_exc()}")

                
                
                if "429" in error_msg :
                    
                    wait_time = 5 * (attempt + 1)
                    logger.warning(f"检测到 429 错误，等待 {wait_time} 秒后重试...")
                    time.sleep(wait_time)
                
                elif attempt < 2:
                    time.sleep(2)
                continue
        return None

多路探测防御性解析

如果我们使用 MCP 使用得多了，就可能会发现 MCP 返回的数据结构不同的厂商可能有所差异。这就导致数据解析中也不可能是一成不变的，为了解决这种工具调用返回结构的”不确定性“问题，这里写了一个函数用于多路探测防御性解析 返回数据。

  • 使用预定义的 JSONPath 列表递归解析

  • 使用抽象归一的解析函数代替分散的if/else代码，几种统一

  • 扩展性友好，后续新增路径只需要更新预定义的 JSONPath 列表即可

    def extract_pages_from_mcp_response(self, response, config=None): """从MCP WebSearch响应中提取pages数据（重构版） Args: response: MCP响应对象 config: 解析路径配置，默认使用预定义的JSONPath列表 """ if response is None: raise Exception("Web搜索结果不正确") if not response.status_code == 200: raise Exception(f"Web搜索异常: {response.status_code}")

    DEFAULT_PARSE_CONFIG = {
    "paths": [
        "$.pages",                  
        "$.data.pages",             
        "$.result.content[0].text",  
        "$.choices[0].message.content",
        
    ]
}

parse_config = config if config else DEFAULT_PARSE_CONFIG

try:
    
    first_level = json.loads(response.output.text)
except json.JSONDecodeError as e:
    logger.error(f"JSON解析失败: {e}, 原始响应: {response.output.text[:500]}")
    raise Exception(f"Web搜索结果JSON解析失败: {str(e)}")

pages = None
matched_path = None


for path_str in parse_config["paths"]:
    try:
        
        jsonpath_expr = jsonpath_parser.parse(path_str)
        matches = [match.value for match in jsonpath_expr.find(first_level)]
        if matches:
            
            candidate = matches[0]

            
            if isinstance(candidate, str):
                try:
                    candidate = json.loads(candidate)
                    
                    
                    if isinstance(candidate, dict) and "pages" in candidate:
                        pages = candidate["pages"]
                        matched_path = path_str + " -> inner.pages"
                        break
                    
                    elif isinstance(candidate, list):
                        pages = candidate
                        matched_path = path_str + " -> inner.list"
                        break
                except json.JSONDecodeError:
                    
                    continue

            
            elif isinstance(candidate, list):
                pages = candidate
                matched_path = path_str
                break

    except JSONPathError as e:
        logger.warning(f"JSONPath语法错误或匹配失败: {path_str}, Error: {e}")
        continue
    except (KeyError, IndexError, TypeError) as e:
        
        continue


if pages is None:
    logger.error(
        f"无法从响应中提取pages数据。尝试的路径: {parse_config['paths']}。"
        f"响应结构: {json.dumps(first_level, ensure_ascii=False)[:500]}"
    )
    raise Exception("无法从Web搜索结果中提取页面数据")

if not isinstance(pages, list):
    logger.error(f"pages不是列表类型: {type(pages)}, 匹配路径: {matched_path}")
    raise Exception("Web搜索结果格式错误")

logger.info(f"成功从路径 '{matched_path}' 提取到 {len(pages)} 条pages数据")


processed_pages = []
for page in pages:
    if isinstance(page, dict):
        processed_pages.append({
            "snippet": page.get("snippet", ""),
            "title": page.get("title", ""),
            "url": page.get("url", "")
        })

return processed_pages

WebSearch MCP

在以上代码我们也发现会用到 MCP_APP_ID 等信息。这是哪里来的呢？这个 id 来源于我们使用的哪个厂商的 MCP 工具。以我们项目为🌰，WebSearch MCP 使用的是阿里百炼平台的网络搜索 MCP。

image.png

image.png

展望

至此，我们完成了Agent 基础的相关代码。包括 LLM 封装，Prompt 渲染，Agent 类封装等。后面有机会，我们继续用 Graph 实现主流程。