上周三凌晨两点我被一个奇怪的问题卡住了团队开发的天气查询Agent在解析某海外城市温度时总是把华氏度当成摄氏度返回。本地测试时明明写死了转换逻辑但线上跑起来就出错。盯着日志看了半小时才反应过来——Agent在回答前偷偷调用了搜索引擎抓到了过时的英文页面数据。这个坑让我彻底明白给Agent接入网络搜索不是简单加个API调用就完事的。今天我们就来拆解这个看似简单、实则暗藏玄机的“信息获取Agent”。为什么需要独立的搜索Agent很多新手喜欢在主Agent里直接塞个requests.get()或者调一下SerpAPI。这做法在Demo里跑得通真上生产就崩。问题出在哪第一网络请求不稳定超时、封IP、页面结构变化都会让主Agent逻辑被打断。第二搜索结果需要清洗网页里广告、导航栏、脚本代码全是噪声。第三——也是最关键的——搜索本身是个策略问题关键词怎么组织用哪个搜索引擎要不要翻页这些决策不应该干扰主Agent的任务推理。所以我们的原则是把搜索抽象成一个独立的工具Agent让它专注做“信息管道”的脏活累活。核心架构三层过滤机制我现在的搜索Agent标配三层处理链代码大概长这样classSearchAgent:def__init__(self):# 第一层查询优化器self.query_rewriterQueryRewriter()# 第二层执行器支持多个搜索引擎降级切换self.executorSearchExecutor()# 第三层内容提取器self.extractorContentExtractor()asyncdefsearch(self,raw_query:str,max_results:int5):# 重写查询词这里踩过坑原样传递用户输入效果很差optimized_queryself.query_rewriter.rewrite(raw_query)# 执行搜索自动降级pages[]forenginein[google,bing,duckduckgo]:try:pagesawaitself.executor.fetch(engine,optimized_query,max_results)ifpages:breakexceptSearchEngineErrorase:logger.warning(f{engine}failed:{e})continue# 提取正文过滤垃圾内容cleaned_contents[]forpageinpages:# 别直接扔原始HTML给LLMtoken浪费严重contentself.extractor.clean_page(page)ifself.extractor.is_high_quality(content):cleaned_contents.append(content[:5000])# 长度截断returncleaned_contents那些容易踩坑的细节查询重写是门玄学早期我直接拿用户问题当搜索词比如用户问“明天杭州穿什么衣服合适”搜回来全是电商广告。现在我会让一个小模型先做查询扩展# 把原始问题转成搜索引擎友好的关键词defrewrite_query(self,raw_query):promptf 请将以下问题转换为2-3个搜索关键词要求 1. 包含具体地点、时间如有 2. 排除主观描述词如“合适”“最好” 3. 优先使用中文关键词 问题{raw_query}关键词 # 调用小模型比如Qwen-2.5-7B生成returnllm_generate(prompt)# 输出示例# 原始“明天杭州穿什么衣服合适”# 改写“杭州明日天气 穿衣指数 气温”降级策略必须有千万别只接一个搜索引擎API。国内环境尤其复杂有些域名访问不稳定。我的经验是至少准备三个后备源按成功率动态调整优先级。有个取巧的办法用多个免费API轮询虽然每个都有额度限制但堆起来够用了。内容提取别迷信通用库readability、newspaper3k这些库对新闻网站效果好但对论坛、文档站经常抽风。我最后维护了一套混合规则先用通用库提取如果提取结果太短比如少于200字就回退到基于标签密度的自研提取器。判断“高质量内容”的规则也很粗暴正文长度300字、广告关键词占比5%、包含至少一个数字或日期。与主Agent的对接模式搜索Agent返回原始文本后怎么交给主Agent使用我试过三种模式直接拼接把搜索结果拼进Prompt。简单但容易超token而且LLM可能分不清哪些是搜索结果、哪些是指令。向量检索把搜索内容向量化存起来主Agent按需检索。这方案理论优雅但延迟增加小项目不值当。摘要模式让搜索Agent先做一次摘要只传摘要过去。这是我目前用的折中方案——摘要用低成本模型生成控制在500字内主Agent需要细节时再按需请求原始片段。# 当前采用的摘要模式示例asyncdefsearch_with_summary(self,query):contentsawaitself.search(query)summaries[]fortextincontents:# 用快模型生成摘要关键要求保留数字、日期、实体名summaryfast_llm.summarize(text,instruction提取关键事实保留数据和时间)summaries.append(f[来源{len(summaries)1}]{summary})# 给主Agent的格式清晰标注来源return\n---\n.join(summaries)自定义工具开发连接外部API与数据库工具不是函数是Agent的感官很多人把自定义工具理解成写个Python函数这是第一个误区。在Agent眼里工具是它感知和操作外部世界的唯一通道。你写的不是代码是给AI的“说明书”。defquery_weather(city:str)-str: 查询指定城市的天气情况 注意这里踩过坑——返回必须是纯文本JSON字符串里别带控制字符 Agent的解析器对格式极其敏感多一个空格都可能解析失败 Args: city: 城市名比如北京、上海 Returns: 格式化的天气信息字符串别返回复杂对象 # 模拟API调用data{city:city,temp:22℃,condition:晴,humidity:65%}# 关键点返回前strip一下血的教训returnjson.dumps(data,ensure_asciiFalse).strip()连接外部API超时是杀手上周帮团队排查一个生产问题Agent调用第三方天气API时卡死拖垮了整个会话。后来发现对方API平均响应要8秒而默认超时只有5秒。importaiohttpimportasynciofromtypingimportOptionalasyncdefcall_external_api(url:str,params:dict,timeout:float10.0)-Optional[dict]: 调用外部API的通用封装 经验之谈 1. 超时设置必须比Agent的超时短否则链式超时 2. 一定要有重试机制但别无限重试 3. 错误信息要友好Agent看不懂HTTP状态码 retry_count0max_retries2whileretry_countmax_retries:try:asyncwithaiohttp.ClientSession()assession:# 这里有个细节timeout用aiohttp.ClientTimeout包装timeout_objaiohttp.ClientTimeout(totaltimeout)asyncwithsession.get(url,paramsparams,timeouttimeout_obj)asresponse:ifresponse.status200:# 注意编码问题有些API返回gbktextawaitresponse.text(encodingutf-8)returnjson.loads(text)else:# 错误信息要结构化方便Agent理解return{error:fAPI返回状态码{response.status},suggestion:请检查城市名称是否正确}exceptasyncio.TimeoutError:retry_count1ifretry_countmax_retries:return{error:请求超时请稍后重试}awaitasyncio.sleep(1)# 简单退避exceptExceptionase:# 别把异常堆栈扔给Agent它看不懂return{error:f系统内部错误:{str(e)}}returnNone数据库连接连接池是必须品见过有人每个工具调用都新建数据库连接结果Agent并发稍高就把数据库连接打满。数据库工具一定要用连接池。importaiomysqlfromcontextlibimportasynccontextmanager# 全局连接池在应用启动时初始化_poolNoneasyncdefinit_db_pool():应用启动时调用一次就行global_pool _poolawaitaiomysql.create_pool(hostlocalhost,port3306,useragent,passwordyour_password,dbagent_db,minsize5,# 最小连接数maxsize20,# 最大连接数按需调整autocommitTrue)asynccontextmanagerasyncdefget_db_connection(): 获取数据库连接的上下文管理器 重要一定要用上下文管理器确保连接归还到池里 我见过内存泄漏就是因为连接没归还 if_poolisNone:raiseRuntimeError(数据库连接池未初始化)connawait_pool.acquire()try:yieldconnfinally:# 这个finally一定要写否则连接泄露_pool.release(conn)asyncdefquery_user_info(user_id:int)-dict: 查询用户信息示例 注意参数一定要做类型检查和转义防止SQL注入 虽然Agent不会故意注入但用户输入不可信 ifnotisinstance(user_id,int):return{error:用户ID必须是数字}asyncwithget_db_connection()asconn:asyncwithconn.cursor(aiomysql.DictCursor)ascursor:# 参数化查询安全第一awaitcursor.execute(SELECT id, name, email FROM users WHERE id %s,(user_id,))resultawaitcursor.fetchone()ifresult:# 数据库字段名可能和Agent期望的不一致return{user_id:result[id],user_name:result[name],contact:result[email]}else:return{error:用户不存在}工具描述Agent的“产品说明书”工具的docstring比代码更重要。Agent通过描述理解工具用途写得太简略Agent会乱用。defsearch_products(keyword:str,max_results:int10)-str: 在商品数据库中搜索产品 这个描述Agent能看到要写清楚 1. 什么情况下用这个工具用户想买东西时 2. 参数具体指什么keyword是商品关键词 3. 返回什么格式JSON列表每个商品包含哪些字段 坏例子搜索商品 —— 太模糊Agent可能误用 好例子当用户询问商品信息、想要购买某类商品时使用此工具 Args: keyword: 商品关键词如智能手机、无线耳机 max_results: 返回结果数量默认10条最多50条 Returns: JSON字符串格式 { products: [ { id: 商品ID, name: 商品名称, price: 价格, in_stock: 是否有货 } ] } # 实现省略...错误处理给Agent明确的指引工具抛异常等于让Agent“瞎了”。所有错误都必须捕获返回结构化错误信息。defsafe_tool_call(func): 工具函数的装饰器统一错误处理 经验不加这个装饰器工具抛异常Agent就卡住了 去年排查一个线上问题就是因为API返回了NoneType asyncdefwrapper(*args,**kwargs):try:resultawaitfunc(*args,**kwargs)# 统一返回格式成功有data字段失败有error字段ifisinstance(result,dict)anderrorinresult:returnresultelse:return{data:result,error:None}exceptjson.JSONDecodeError:return{error:数据解析失败请稍后重试}exceptKeyErrorase:return{error:f数据缺少必要字段:{str(e)}}exceptExceptionase:# 生产环境这里要打日志但别返回给Agentlogger.error(f工具{func.__name__}执行失败:{e})return{error:系统繁忙请稍后再试}returnwrapper几个实战建议工具粒度要适中一个工具做一件事。见过有人把“查询天气预订机票”做在一个工具里Agent完全不会用。参数设计要直观Agent理解不了复杂参数。比如时间参数接受“明天”、“下周”这种自然语言在工具内部转成具体日期。返回数据要精简Agent的上下文长度有限。数据库查询只返回必要字段别SELECT *。工具要有状态感知需要登录的工具先检查会话状态。见过没做检查的工具Agent在未登录状态下调用了需要权限的API。版本兼容要考虑工具接口一旦发布不要轻易修改。必须修改时保持向后兼容至少两个版本。监控必须加每个工具调用都要打点记录耗时、成功率。我们曾经有个工具平均耗时从200ms慢慢涨到2s没监控根本发现不了。最后说个真事团队里有个工程师写的工具一直工作正常直到用户输入了emoji表情。数据库字段是utf8mb3存不了四字节的utf8整个调用链崩溃。现在我们的第一条军规就是所有字符串输入先做清洗所有数据库连接都设成utf8mb4。工具开发就像给AI造手和眼睛——手要灵活可靠眼睛要看得清楚。代码写得好不好看Agent用得顺不顺手就知道。