Langchain Agent实战避坑：用通义千问调用高德API，我踩过的异步和工具定义那些坑

Langchain Agent实战避坑指南异步调用与工具定义的那些坑在构建基于Langchain的智能代理时异步调用和工具定义是两个最容易让开发者踩坑的领域。本文将分享我在使用通义千问模型调用高德API过程中遇到的实际问题及其解决方案。1. 异步调用的常见陷阱异步编程在Langchain Agent中的应用看似简单实则暗藏玄机。以下是几个典型的异步调用问题1.1 异步方法未被正确触发许多开发者按照文档实现了_arun方法却发现Agent始终调用同步的_run方法。这通常是由于工具初始化方式不当造成的# 错误示例直接实例化工具类 tool WeatherTool() # 这样初始化会导致异步调用失效 # 正确做法通过Tool包装器注册 from langchain.tools import Tool weather_tool Tool( nameget_weather, funcweather_tool._run, coroutineweather_tool._arun, # 关键显式指定协程方法 description获取指定地点的当前天气 )1.2 异步上下文管理不当使用aiohttp进行异步HTTP调用时常见的错误是未正确管理ClientSession生命周期# 错误示例每次调用都创建新session async def _arun(self, location: str): async with aiohttp.ClientSession() as session: # 频繁创建销毁影响性能 # ... # 推荐方案复用session class WeatherTool(BaseTool): def __init__(self): self.session None async def _arun(self, location: str): if not self.session: self.session aiohttp.ClientSession() # 使用现有session...1.3 异步任务编排问题当需要并行调用多个工具时新手常犯的错误是顺序执行而非真正并行# 低效做法顺序等待 result1 await tool1.arun(input1) result2 await tool2.arun(input2) # 必须等tool1完成 # 高效方案并行执行 import asyncio async def gather_results(): tasks [ tool1.arun(input1), tool2.arun(input2) ] return await asyncio.gather(*tasks) # 同时触发所有任务2. 工具定义的进阶技巧工具定义的质量直接影响Agent的决策能力。以下是几个关键优化点2.1 描述字段的黄金法则工具描述(description)不是注释而是Agent选择工具的关键依据。优质描述应包含明确的作用范围具体说明工具处理什么类型的问题输入格式要求指定参数格式和单位输出内容说明告知Agent会得到什么样的信息# 普通描述 description获取城市天气 # 优化描述 description( 获取中国城市实时天气数据。输入应为标准城市名称如北京市 输出包含天气状况晴/雨等和温度摄氏度。 仅支持地级市及以上城市查询。 )2.2 参数验证的最佳实践Pydantic模型可以大幅提升工具的鲁棒性from pydantic import BaseModel, Field, validator class WeatherInput(BaseModel): location: str Field(..., description城市名称如上海市, examples[北京市, 广州市]) validator(location) def check_city(cls, v): if v not in CITY_ADMAPPING: raise ValueError(f不支持的城市: {v}) return v class WeatherTool(BaseTool): args_schema WeatherInput # 绑定参数验证模型 # ...2.3 多工具协作的命名策略当Agent需要处理多个相似工具时命名要有区分度工具类型反例命名推荐命名优势天气查询get_dataget_weather_by_city明确功能范围人口查询get_dataget_population_stats避免歧义地理编码convertgeocode_location体现专业术语3. 高德API集成特别注意事项对接高德天气API时这些细节容易忽略3.1 城市编码映射的优化直接使用城市名查询可能导致API调用失败# 脆弱实现 adcode CITY_ADMAPPING[location] # 可能KeyError # 健壮方案 def normalize_city_name(name: str) - str: name name.replace(市, ).replace(省, ) return CITY_ADMAPPING.get(name) # 支持多种输入格式 normalize_city_name(北京) 110000 normalize_city_name(北京市) 1100003.2 API响应缓存机制为避免频繁调用API建议添加缓存层from functools import lru_cache import time lru_cache(maxsize100) def get_weather_with_cache(adcode: str): # 添加请求时间戳避免缓存命中率过低 params { city: adcode, key: API_KEY, extensions: base, _t: int(time.time() / 300) # 5分钟缓存窗口 } # ...3.3 错误处理的完整模式全面的错误处理应包括async def _arun(self, location: str): try: adcode self.normalize_city_name(location) if not adcode: return f无法识别城市: {location} async with self.session.get(API_URL, paramsparams) as resp: if resp.status ! 200: return 服务暂时不可用 data await resp.json() if data[status] ! 1: return fAPI错误: {data.get(info, 未知)} return self._format_weather(data) except aiohttp.ClientError as e: return f网络错误: {str(e)} except Exception as e: return f处理异常: {str(e)}4. 调试与性能优化当Agent行为不符合预期时这些调试技巧很实用4.1 启用详细日志agent initialize_agent( tools, llmtongyi_chat, verboseTrue, # 打印完整决策链 handle_parsing_errorsTrue # 捕获解析异常 )4.2 思维链(Chain-of-Thought)分析在工具描述中添加推理提示description( 当需要比较两个城市的气候条件时使用本工具。 先分别查询两个城市的天气然后比较温度数值。 输出格式X市比Y市高/低Z摄氏度 )4.3 性能监控装饰器跟踪工具执行耗时import time from functools import wraps def timing_decorator(func): wraps(func) async def wrapper(*args, **kwargs): start time.perf_counter() result await func(*args, **kwargs) elapsed (time.perf_counter() - start) * 1000 print(f{func.__name__} took {elapsed:.2f}ms) return result return wrapper class WeatherTool(BaseTool): timing_decorator async def _arun(self, location: str): # ...5. 实战中的经验之谈在真实项目中这些经验往往能节省大量时间版本兼容性Langchain的异步接口在不同版本间可能有差异建议锁定版本号重试机制对于不稳定的API添加指数退避重试逻辑测试策略不仅测试正常流程更要模拟网络延迟、API限流等异常情况文档注释为每个工具添加完整的docstring方便后续维护def load_city_mapping(file_path: str) - dict: 加载高德城市编码映射表 参数 file_path: CSV文件路径需包含中文名和adcode列 返回 城市名到adcode的映射字典 示例 load_city_mapping(AMap_adcode.csv) {北京: 110000, 上海: 310000} # 实现代码...最终实现的Agent应该能够优雅地处理各种边界情况比如当用户询问比较重庆和天津的人口和天气时可以并行执行四个工具调用两地的天气和人口查询然后综合所有结果生成自然语言响应。