Langchain + 通义千问：打造你的第一个多工具智能体

1. 为什么需要Langchain与通义千问的结合在AI技术快速发展的今天单一模型已经很难满足复杂场景的需求。想象一下你有一个非常聪明的助手但它只会回答问题却不会帮你查天气、查航班或者处理其他具体事务。这就是为什么我们需要Langchain这样的框架——它就像是一个万能工具箱可以把各种专用工具和强大的语言模型连接在一起。通义千问作为国内领先的大语言模型在中文理解和生成方面表现出色。但单独使用时它更像是一个知识渊博的学者知道很多却不会实际操作。而Langchain则像是给这位学者配了一个助理团队当学者需要具体数据时助理们可以立即去查询、计算或处理。我最近在做一个智能客服项目时就深有体会。客户不仅需要回答问题还经常询问上海明天天气如何、北京人口有多少这类需要实时数据的问题。单纯依赖语言模型根本无法满足需求直到我发现了Langchain这个神器。2. 环境准备与依赖安装2.1 基础环境配置在开始之前确保你的Python版本在3.8以上。我推荐使用虚拟环境来管理依赖这样可以避免包冲突。创建并激活虚拟环境的命令如下python -m venv langchain-env source langchain-env/bin/activate # Linux/Mac # 或者 langchain-env\Scripts\activate # Windows接下来安装核心依赖。除了原始文章中提到的包我还建议安装几个非常有用的附加工具pip install requests langchain langchain-community pydantic aiohttp python-dotenv这里解释下每个包的作用requests处理HTTP请求的瑞士军刀langchain核心框架langchain-community社区贡献的扩展包括通义千问集成pydantic数据验证神器aiohttp异步HTTP客户端python-dotenv管理环境变量更安全2.2 获取API密钥你需要准备两个API密钥通义千问API密钥从阿里云百炼平台获取高德地图API密钥用于天气查询建议将密钥存储在环境变量中而不是直接写在代码里。创建一个.env文件DASHSCOPE_API_KEY你的通义千问密钥 GAODE_API_KEY你的高德地图密钥然后在代码开头加载这些变量from dotenv import load_dotenv load_dotenv() import os DASHSCOPE_API_KEY os.getenv(DASHSCOPE_API_KEY) GAODE_API_KEY os.getenv(GAODE_API_KEY)3. 定义你的第一个工具天气查询3.1 城市编码映射天气查询需要将城市名称转换为高德地图的adcode。我改进了一下原始代码中的映射加载方式增加了错误处理def load_city_mapping(file_path): try: with open(file_path, moder, encodingutf-8) as file: return {row[中文名]: row[adcode] for row in csv.DictReader(file) if row[中文名] and row[adcode]} except FileNotFoundError: print(f错误找不到文件 {file_path}) return {} except Exception as e: print(f加载映射表时出错{str(e)}) return {}3.2 增强版天气工具原始代码的天气工具已经很实用但我增加了缓存功能避免频繁调用APIfrom datetime import datetime, timedelta class WeatherTool(BaseTool): name get_weather description 获取指定地点的当前天气 def __init__(self): super().__init__() self.cache {} self.cache_time timedelta(minutes30) # 缓存30分钟 def _get_from_cache(self, location): item self.cache.get(location) if item and datetime.now() - item[time] self.cache_time: return item[data] return None def _run(self, location: str): # 先检查缓存 cached self._get_from_cache(location) if cached: return cached adcode CITY_MAPPING.get(location) if not adcode: return f无法找到{location}对应的城市编码 params { city: adcode, key: GAODE_API_KEY, extensions: base } try: response requests.get(GAODE_WEATHER_URL, paramsparams, timeout5) data response.json() if data[status] 1: weather data[lives][0] result f{location}天气{weather[weather]}温度{weather[temperature]}℃湿度{weather[humidity]}%风向{weather[winddirection]}风力{weather[windpower]}级 # 存入缓存 self.cache[location] {data: result, time: datetime.now()} return result else: return f获取天气失败{data.get(info, 未知错误)} except requests.exceptions.RequestException as e: return f网络请求出错{str(e)}4. 创建智能体并初始化模型4.1 初始化通义千问模型Langchain社区版已经集成了通义千问初始化非常简单from langchain_community.chat_models.tongyi import ChatTongyi tongyi_chat ChatTongyi( api_keyDASHSCOPE_API_KEY, modelqwen-plus, # 也可以尝试qwen-max获得更强能力 temperature0.7, # 控制创造性0-1之间 top_p0.9 # 核采样参数 )4.2 组装工具集除了天气工具我们再创建一个更实用的工具 - 城市信息查询class CityInfoTool(BaseTool): name get_city_info description 获取城市的基本信息包括GDP、面积等 def _run(self, location: str): # 这里应该是调用城市信息API # 为演示我们返回模拟数据 info { GDP: f{random.randint(100, 5000)}亿元, 面积: f{random.randint(1000, 20000)}平方公里, 行政区划代码: CITY_MAPPING.get(location, 未知) } return \n.join(f{k}: {v} for k,v in info.items())4.3 创建智能体将工具和模型组合起来from langchain.agents import AgentType, initialize_agent tools [ Tool.from_function( funcWeatherTool()._run, nameget_weather, description获取指定城市的当前天气 ), Tool.from_function( funcCityInfoTool()._run, nameget_city_info, description获取城市基本信息如GDP、面积等 ) ] agent initialize_agent( tools, tongyi_chat, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verboseTrue # 打印详细执行过程 )5. 异步调用与性能优化5.1 为什么要用异步当智能体需要同时查询多个信息源时同步调用会导致不必要的等待。异步可以让多个IO操作并行执行显著提升响应速度。5.2 实现异步工具改进天气工具实现真正的异步查询class AsyncWeatherTool(BaseTool): name get_weather_async description 异步获取天气信息 async def _arun(self, location: str): adcode CITY_MAPPING.get(location) if not adcode: return f无法找到{location}对应的城市编码 params { city: adcode, key: GAODE_API_KEY, extensions: all } try: async with aiohttp.ClientSession() as session: async with session.get(GAODE_WEATHER_URL, paramsparams) as resp: data await resp.json() if data[status] 1: weather data[lives][0] return (f{location}天气{weather[weather]} f温度{weather[temperature]}℃) else: return f获取天气失败{data.get(info, 未知错误)} except Exception as e: return f请求出错{str(e)}5.3 异步执行智能体async def async_main(): async_tools [ Tool.from_function( funcAsyncWeatherTool()._arun, nameget_weather, description获取指定城市的当前天气 ), # 其他异步工具... ] async_agent initialize_agent( async_tools, tongyi_chat, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verboseTrue ) response await async_agent.arun(比较北京和上海的人口、GDP和当前天气) print(最终结果:, response)6. 实际应用案例与调试技巧6.1 典型问题排查在开发过程中我遇到过几个常见问题工具未被调用检查description是否准确描述了工具功能LLM会根据描述决定是否调用API限流添加适当的延迟和重试机制解析错误确保工具返回的数据格式与LLM期望的一致6.2 增强提示工程通过优化提示词可以获得更好的结果from langchain.prompts import MessagesPlaceholder from langchain.memory import ConversationBufferMemory memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue) agent_kwargs { extra_prompt_messages: [MessagesPlaceholder(variable_namechat_history)], system_message: 你是一个专业的城市信息助手能够查询天气、人口、经济数据等。回答要专业且友好。 } agent initialize_agent( tools, tongyi_chat, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, memorymemory, agent_kwargsagent_kwargs, verboseTrue )6.3 实际应用示例# 查询多个城市比较 question 请比较北京、上海和广州的以下信息 1. 当前天气状况 2. 人口规模 3. GDP数据 用表格形式展示并给出总结分析 response agent.run(question) print(response)这个查询会触发智能体自动调用相应的工具获取数据然后由通义千问整理成结构化的回答。