Langchain 和通义千问作为现代 AI 工具链中的佼佼者，正在为开发者提供前所未有的便利。本篇博客将通过一个简单的示例，结合实际代码，介绍如何使用 Langchain 来构建一个多工具集成的智能代理（Agent）。我将带领大家实现一个智能助手，能够根据用户输入自动调用多个工具，获取城市的天气和人口信息。

1. 环境准备

在开始之前，请确保你已经安装了以下依赖：

  1. requests：用于发送 HTTP 请求，访问外部 API。

    pip install requests

  2. langchain：用于处理和组织不同的工具、模型以及代理。

    pip install langchain

  3. langchain_community：用于使用通义千问模型，这个包是 Langchain 社区提供的扩展。

    pip install langchain_community

  4. pydantic：用于数据验证和序列化，确保输入输出数据格式正确。

    pip install pydantic

  5. aiohttp：由于我们在示例中使用了异步编程来进行 HTTP 请求，aiohttp 是必需的。

    pip install aiohttp

  6. utils：这个工具模块中包含了加载城市编码映射的函数。

    如果你没有 utils.py 文件，下面是示例，来加载城市编码映射：

    import csv

# 从 CSV 文件中加载城市名称到 adcode 的映射
def load_city_adcode_mapping(file_path):
    mapping = {}
    with open(file_path, mode='r', encoding='utf-8') as file:
        reader = csv.DictReader(file)
        for row in reader:
            city_name = row['中文名']
            adcode = row['adcode']
            if city_name and adcode:
                mapping[city_name] = adcode
    return mapping

 

你还需要获取以下 API 密钥：

2. 定义工具

在 Langchain 中，工具是实现特定功能的核心。我们将定义两个工具：

  1. 天气工具：调用高德地图 API 获取指定城市的天气信息。
  2. 人口工具：我通过随机数生成器模拟返回城市的人口数量。如果要使用API来获得精准的城市人数的话更佳，代码编写可参照天气工具(根据相应API文档编写POST/GET请求并处理返回值）
2.1 天气工具（WeatherTool）
# 将工具包装为 langchain 的 BaseTool
class WeatherTool(BaseTool):
    name: str = "get_weather"
    description: str = "获取指定地点的当前天气"

    def _run(self, location: str):
        # 将城市名称映射为 adcode
        adcode = CITY_ADMAPPING.get(location)
        if not adcode:
            return f"'{location}' 未能在映射表中找到相应映射"

        # 调用高德地图 API 获取天气数据
        params = {
            "city": adcode,
            "key": GAODE_API_KEY,
            "extensions": "base"  # 获取实况天气信息
        }
        response = requests.get(GAODE_WEATHER_URL, params=params)
        data = response.json()
        if data["status"] == "1":
            weather_info = data["lives"][0]["weather"]
            temperature = data["lives"][0]["temperature"]
            return f" {location}的天气为{weather_info}，温度为{temperature}"
        else:
            return f"获得{location}该地天气信息失败: {data['info']}"

    async def _arun(self, location: str):
        # 异步版本
        import aiohttp
        adcode = CITY_ADMAPPING.get(location)
        if not adcode:
            return f"'{location}' 未能在映射表中找到相应映射"

        async with aiohttp.ClientSession() as session:
            async with session.get(GAODE_WEATHER_URL, params={
                "city": adcode,
                "key": GAODE_API_KEY,
                "extensions": "all"
            }) as response:
                data = await response.json()
                if data["status"] == "1":
                    weather_info = data["lives"][0]["weather"]
                    temperature = data["lives"][0]["temperature"]
                    return f"{location}的天气为{weather_info}，温度为{temperature}"
                else:
                    return f"获得{location}该地天气信息失败: {data['info']}"
2.2 人口工具（PopulationTool）
class PopulationTool(BaseTool):
    name: str = "get_population"
    description: str = "获取指定地点的当前人口数量"
    
    def _run(self, location: str):
        # 生成一个介于 100 到 5000 之间的随机数来模拟API返回数据
        population = random.randint(100, 5000)
        return f"{location}当地人口为{population}万"

    async def _arun(self, location: str):
        # 异步版本
        # 生成一个介于 100 到 5000 之间的随机数来模拟API返回数据
        population = random.randint(100, 5000)
        return f"{location}当地人口为{population}万"

3. 创建 Agent 并初始化模型

通过 Langchain，可以将这些工具与一个强大的语言模型（本次使用通义千问的模型：qwen-plus）结合起来，创建一个可以处理多步推理的智能代理。

# 初始化工具
weather_tool = WeatherTool()
population_tool = PopulationTool()

# 创建工具列表
tools = [
    Tool(
        name="get_weather",
        func=weather_tool._run,
        description="获取指定地点的当前天气"
    ),
    Tool(
        name="get_population",
        func=population_tool._run,
        description="获取指定地点的当前人口数量"
    )
]

# 初始化 Agent，将工具绑定到模型
agent = initialize_agent(tools, llm=tongyi_chat, verbose=True)

4. 异步调用 Agent

在实际应用中，我们希望能够并发执行多个操作。因此，我们使用异步编程来并行获取天气和人口信息。以下是如何通过异步函数调用 Langchain Agent：

# 定义异步函数来调用 Agent
async def main():
    # 调用 Agent 并传入问题
    response = await agent.ainvoke("哪个城市今天更热，哪个城市人更多: 成都市 还是 长春市?")
    print("Response:", response)

# 运行异步主函数
if __name__ == "__main__":
    asyncio.run(main())

5. 运行结果

当我们运行上面的代码时，Langchain Agent 会根据用户的输入自动调用天气和人口工具，并生成思维链以及最终的返回值，下面是一次典型输出：

> Entering new AgentExecutor chain...
Action: get_weather
Action Input: 成都市
Observation: 成都市的天气为阴，温度为9
Thought:我已获取成都市的天气信息，接下来需要获取长春市的天气信息。
Action: get_weather
Action Input: 长春市
Observation: 长春市的天气为晴，温度为3
Thought:我已获取长春市的天气信息，现在需要获取这两个城市的人口数量信息。
Action: get_population
Action Input: 成都市
Observation: 成都市当地人口为1746万
Thought:我已获取成都市的人口信息，现在需要获取长春市的人口信息。
Action: get_population
Action Input: 长春市
Observation: 长春市当地人口为3630万
Thought:我已获取所有需要的信息，现在可以进行比较并得出结论。
Final Answer: 今天成都市比长春市更热，而长春市的人口比成都市更多。

> Finished chain.
Response: {'input': '哪个城市今天更热，哪个城市人更多: 成都市 还是 长春市?', 'output': '今天成都市比长春市更热，而长春市的人口比成都市更多。'}

6. 总结

通过本例，我们展示了如何使用 Langchain 和通义千问模型来构建一个具备多工具集成的智能代理。该代理能够基于用户问题自动调用相关工具获取信息，并生成合理的答案。Langchain 作为一种灵活强大的框架，不仅支持多种工具和模型的集成，还支持异步调用，让构建复杂的智能系统变得更加简单和高效。

希望通过这个小示例，你能更好地理解 Langchain 的强大功能，未来也可以尝试构建更多有趣的应用。感谢你的观看，我是Tex-mind，我们下次再见！

7. 完整代码和相关资源

完整代码如下：
app.py 👇

import requests
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain.tools import BaseTool
from langchain.agents import initialize_agent, Tool
from langchain.schema import AIMessage
import random
from pydantic import BaseModel, Field
import asyncio
from utils import (
    load_city_adcode_mapping
)

# 设置你的 DashScope API Key 和模型
DASHSCOPE_API_KEY = ""
DASHSCOPE_MODEL = "qwen-plus"

# 初始化 ChatTongyi 模型
tongyi_chat = ChatTongyi(
    api_key=DASHSCOPE_API_KEY,
    model=DASHSCOPE_MODEL
)

# 高德地图 API 配置
GAODE_API_KEY = ""
GAODE_WEATHER_URL = "https://restapi.amap.com/v3/weather/weatherInfo"
# 加载映射表
CITY_ADMAPPING = load_city_adcode_mapping("AMap_adcode_citycode.csv")


# 定义获取天气的工具
class GetWeather(BaseModel):
    '''获取指定地点的当前天气'''
    location: str = Field(..., description="城市名称，例如：北京市")

# 定义获取人口的工具
class GetPopulation(BaseModel):
    '''获取指定地点的当前人口数量'''
    location: str = Field(..., description="城市名称，例如：北京市")

# 将工具包装为 langchain 的 BaseTool
class WeatherTool(BaseTool):
    name: str = "get_weather"
    description: str = "获取指定地点的当前天气"

    def _run(self, location: str):
        # 将城市名称映射为 adcode
        adcode = CITY_ADMAPPING.get(location)
        if not adcode:
            return f"'{location}' 未能在映射表中找到相应映射"

        # 调用高德地图 API 获取天气数据
        params = {
            "city": adcode,
            "key": GAODE_API_KEY,
            "extensions": "base"  # 获取实况天气信息
        }
        response = requests.get(GAODE_WEATHER_URL, params=params)
        data = response.json()
        if data["status"] == "1":
            weather_info = data["lives"][0]["weather"]
            temperature = data["lives"][0]["temperature"]
            return f" {location}的天气为{weather_info}，温度为{temperature}"
        else:
            return f"获得{location}该地天气信息失败: {data['info']}"

    async def _arun(self, location: str):
        # 异步版本
        import aiohttp
        adcode = CITY_ADMAPPING.get(location)
        if not adcode:
            return f"'{location}' 未能在映射表中找到相应映射"

        async with aiohttp.ClientSession() as session:
            async with session.get(GAODE_WEATHER_URL, params={
                "city": adcode,
                "key": GAODE_API_KEY,
                "extensions": "all"
            }) as response:
                data = await response.json()
                if data["status"] == "1":
                    weather_info = data["lives"][0]["weather"]
                    temperature = data["lives"][0]["temperature"]
                    return f"{location}的天气为{weather_info}，温度为{temperature}"
                else:
                    return f"获得{location}该地天气信息失败: {data['info']}"


class PopulationTool(BaseTool):
    name: str = "get_population"
    description: str = "获取指定地点的当前人口数量"
    
    def _run(self, location: str):
        # 生成一个介于 100 到 5000 之间的随机数来模拟API返回数据
        population = random.randint(100, 5000)
        return f"{location}当地人口为{population}万"

    async def _arun(self, location: str):
        # 异步版本
        # 生成一个介于 100 到 5000 之间的随机数来模拟API返回数据
        population = random.randint(100, 5000)
        return f"{location}当地人口为{population}万"

# 初始化工具
weather_tool = WeatherTool()
population_tool = PopulationTool()

# 创建工具列表
tools = [
    Tool(
        name="get_weather",
        func=weather_tool._run,
        description="获取指定地点的当前天气"
    ),
    Tool(
        name="get_population",
        func=population_tool._run,
        description="获取指定地点的当前人口数量"
    )
]

# 初始化 Agent，将工具绑定到模型
agent = initialize_agent(tools, llm=tongyi_chat, verbose=True)

# 定义异步函数来调用 Agent
async def main():
    # 调用 Agent 并传入问题
    response = await agent.ainvoke("哪个城市今天更热，哪个城市人更多: 成都市 还是 长春市?")
    print("Response:", response)

# 运行异步主函数
if __name__ == "__main__":
    asyncio.run(main())

utils.py👇

import csv

# 从 CSV 文件中加载城市名称到 adcode 的映射
def load_city_adcode_mapping(file_path):
    mapping = {}
    with open(file_path, mode='r', encoding='utf-8') as file:
        reader = csv.DictReader(file)
        for row in reader:
            city_name = row['中文名']
            adcode = row['adcode']
            if city_name and adcode:
                mapping[city_name] = adcode
    return mapping

程序使用的城市名称和代码映射表 AMap_adcode_citycode.csv 👇

相关下载-Web服务 API | 高德地图API

高德地图天气API调用官方教程👇

天气查询-基础 API 文档-开发指南-Web服务 API | 高德地图API

# python # 开发语言
Logo
DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

cover

RoboTTT：机器人策略的上下文规模化

avatar DAMO开发者矩阵

开题季实测：AI能帮你写报告，但大模型别直接交

开题季自救指南：测了十几款AI工具后，我劝你别再拿通用大模型硬写论文了 开学没几天，后台问"开题报告怎么写"的消息又炸了。 说个真事：我同门上周用某知名大模型生成开题报告，参考文献列了12篇，导师随手一查——7篇搜不到，剩下5篇里有3篇年份和作者对不上。当场被打回，还落了个"学术态度不端"的印象。 这两年AI工具爆发，通用大模型、垂直写作平台、AI智能体满天飞，个个都说自己能写论文。但开题报告这东

avatar DAMO开发者矩阵
cover

EgoEngine：从以自我为中心人类视频到高保真灵巧机器人演示

avatar DAMO开发者矩阵