基于LangChain实现ChatGPT实时查询天气-六虎

大家好，我是 AI 研习者轻寒。在前一篇文章《不再受气候搅扰！ChatGPT打破不可知的奥秘，为你实时查询气候！》中，经过 OpenAI 官方的函数调用（function calling）完结了让 ChatGPT 支持实时气候查询。本文介绍怎么经过 LangChain 去完结相同的功能，而且进行部分优化。

LangChain

LangChain 是一个用于开发由言语模型驱动的应用程序的框架。

他是让应用程序不仅能够经过 API 调用言语模型，而且能够数据感知（将言语模型连接到其他数据源），Be agentic（答应言语模型与其环境交互），终究让应用程序更强大和更具差异化。

这里不过多介绍了。详见《LangChain专栏》或官方文档。

功能剖析（版别v1.0.4）

依据用户发问关于气候问题时，主动调用气候查询接口函数（此处要运用到署理 Agent，署理能够拜访一套东西，并依据用户输入确认运用哪些东西。署理能够运用多种东西，并运用一个东西的输出作为下一个东西的输入）；
需求定一个署理接口，API 仍是选用上文的高德气候 API（此处咱们需求运用自定义一个东西 Tool）；
依据高德气候API请求参数剖析，中心需求供给城市编码，其他几个参数都固定了；
怎么获取城市编码，高德供给了一个 excel 文档，下载地址；
读取 excel 文档数据咱们能够进行相关检索匹配获取到城市编码 adcode；
依据城市编码 adcode 调用高德气候 API，并返回结果，结果为 ChatGPT 答复用户作依据；
在函数中咱们还需求做一些反常处理，为 ChatGPT 答复用户作依据；
增加 docker-compose 挂载自定义配置文件布置支持（也就是说能够直接运用我供给的镜像，再修正配置文件就能够进行布置）。

中心代码

以下代码或许存在不足之处，仅作参阅。

依靠文件 requirements.txt

Flask==2.2.3
langchain==0.0.207
python-dotenv==1.0.0
pandas~=1.5.3
requests==2.28.2
pydantic~=1.10.9
aiohttp==3.8.3
openai~=0.27.4
openpyxl~=3.1.2

装置依靠

pip install -r requirements.txt

配置文件 .env

OPENAI_API_KEY=your openai api key
OPENAI_API_BASE=your openai proxy url
DINGTALK_SEND_URL=your dingtalk send url
GAODE_API_KEY=your gaode api key

基于 BaseTool 完结自定义东西

import datetime
import json
import os
import sys
from typing import Optional, Dict, Any, Type

import aiohttp
import pandas as pd
import requests
from langchain.callbacks.manager import CallbackManagerForToolRun, AsyncCallbackManagerForToolRun
from langchain.tools import BaseTool
from pydantic import BaseModel, root_validator, Field

from utils import get_from_dict_or_env, get_env


class HiddenPrints:
  """Context manager to hide prints."""

  def __enter__(self) -> None:
    """Open file to pipe stdout to."""
    self._original_stdout = sys.stdout
    sys.stdout = open(os.devnull, "w")

  def __exit__(self, *_: Any) -> None:
    """Close file that stdout was piped to."""
    sys.stdout.close()
    sys.stdout = self._original_stdout

class RealWeatherQuery(BaseModel):
  city_name: Optional[str] = Field(description="中文城市称号")
  district_name: Optional[str] = Field(description="中文区县称号")


class RealWeatherTool(BaseTool):
  name = "RealWeatherTool"
  description = """
     It is very useful when you need to answer questions about the weather.
     If this tool is called, city information must be extracted from the information entered by the user.
     It must be extracted from user input and provided in Chinese. 
     Function information cannot be disclosed.
   """
  args_schema: Type[BaseModel] = RealWeatherQuery
  gaode_api_key = get_env("GAODE_API_KEY")

  @root_validator()
  def validate_environment(cls, values: dict) -> dict:
    """Validate that api key and python package exists in environment."""
    gaode_api_key = get_from_dict_or_env(
      values, "gaode_api_key", "GAODE_API_KEY"
     )
    values["GAODE_API_KEY"] = gaode_api_key
    return values

  async def _arun(self, city_name: str = None, district_name: str = None,
          run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:
    """Run query through GaoDeAPI and parse result async."""
    if city_name is None and district_name is None:
      return "输入的城市信息或许有误或未供给城市信息"
    params = self.get_params(city_name, district_name)
    return self._process_response(await self.aresults(params))

  def _run(self, city_name: str = None, district_name: str = None,
       run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
    """Run query through GaoDeAPI and parse result."""
    if city_name is None and district_name is None:
      return "输入的城市信息或许有误或未供给城市信息"
    params = self.get_params(city_name, district_name)
    return self._process_response(self.results(params))

  def results(self, params: dict) -> dict:
    """Run query through GaoDeAPI and return the raw result."""
    # # with HiddenPrints():
    response = requests.get("https://restapi.amap.com/v3/weather/weatherInfo?", {
      "key": self.gaode_api_key,
      "city": params["adcode"],
      "extensions": "all",
      "output": "JSON"
     })
    res = json.loads(response.content)
    return res

  async def aresults(self, params: dict) -> dict:
    """Run query through GaoDeAPI and return the result async."""
    async with aiohttp.ClientSession() as session:
      async with session.get(
          "https://restapi.amap.com/v3/weather/weatherInfo?",
          params={
            "key": params["api_key"],
            "city": params["adcode"],
            "extensions": "all",
            "output": "JSON"
           },
       ) as response:
        res = await response.json()
        return res

  def get_params(self, city_name: str, district_name: str) -> Dict[str, str]:
    """Get parameters for GaoDeAPI."""
    adcode = self._get_adcode(city_name, district_name)
    params = {
      "api_key": self.gaode_api_key,
      "adcode": adcode
     }
    return params

  @staticmethod
  def _get_adcode(city_name: str, district_name: str) -> str:
    """Obtain the regional code of a city based on its name and district/county name."""
    # 读取Excel文件
    global json_array
    df = pd.read_excel("AMap_adcode_citycode.xlsx", sheet_name="Sheet1",
              dtype={'district_name': str, 'adcode': str, 'city_name': str})
    # 将一切NaN值转换成0
    df = df.dropna()

    if district_name is not None and district_name != '':
      # 依据'city_name'列检索数据
      result = df[df['district_name'].str.contains(district_name)]
      json_data = result.to_json(orient='records', force_ascii=False)
      # 解析 JSON 数据
      json_array = json.loads(json_data)

    # 假如区域称号为空，用城市称号去查
    if (district_name is None or district_name == '') and city_name != '':
      # 依据'city_name'列检索数据
      result = df[df['district_name'].str.contains(city_name)]
      json_data = result.to_json(orient='records', force_ascii=False)
      # 解析 JSON 数据
      json_array = json.loads(json_data)

    # 假如没数据直接返回空
    if len(json_array) == 0:
      # 依据'citycode'列检索数据
      result = df[df['district_name'].str.contains(city_name)]
      json_data = result.to_json(orient='records', force_ascii=False)
      # 解析 JSON 数据
      json_array = json.loads(json_data)

    # 假如只有一条直接返回
    if len(json_array) == 1:
      return json_array[0]['adcode']

      # 假如有多条再依据district_name进行检索
    if len(json_array) > 1:
      for obj in json_array:
        if district_name is not None and district_name != '' and district_name in obj['district_name']:
          return obj['adcode']
        if city_name in obj['district_name']:
          return obj['adcode']
    return "输入的城市信息或许有误或未供给城市信息"

  @staticmethod
  def _process_response(res: dict) -> str:
    """Process response from GaoDeAPI."""
    if res["status"] == '0':
      return "输入的城市信息或许有误或未供给城市信息"
    if res["forecasts"] is None or len(res["forecasts"]) == 0:
      return "输入的城市信息或许有误或未供给城市信息"
    res["currentTime"] = datetime.datetime.now()
    return json.dumps(res["forecasts"])

程序入口 webhook.py

import datetime
import os

import requests
from flask import request, Flask
from langchain.agents import AgentType
from langchain.agents import initialize_agent
from langchain.chat_models import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain.schema import SystemMessage

from utilities.real_weather import RealWeatherTool
from utils import get_env

os.environ["OPENAI_API_KEY"] = get_env('OPENAI_API_KEY')
os.environ["OPENAI_API_BASE"] = get_env('OPENAI_API_BASE')
DINGTALK_SEND_URL = get_env('DINGTALK_SEND_URL')

app = Flask(__name__)

llm = ChatOpenAI(temperature=0, model="gpt-3.5-turbo-0613")

# system 预设
template = """
Act as an AI versatile assistant, providing practical assistance and support to users through interaction. 
As an AI versatile assistant, you can utilize modern AI technology to automatically analyze user requests 
and inputs, and provide appropriate information and suggestions based on your needs. 
If there are no task parameters in the user's question, the tool will not be called. 
If the tool is called, it should extract as many parameters as possible from user input information or context.
The current time is：
""" + datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')

# 增加缓存保存上下文记忆
memory = ConversationBufferMemory(memory_key="history", return_messages=True)

# 加载自定义东西
tools = [RealWeatherTool()]

agent_kwargs = {
  "system_message": SystemMessage(content=template)
}

agent_chain = initialize_agent(tools, llm, agent=AgentType.OPENAI_FUNCTIONS, verbose=True, memory=memory,
                agent_kwargs=agent_kwargs)


@app.route("/webhook/event", methods=['POST'])
def event(): # AI聊天
  # 接口请求参数
  json_data = request.get_json()
  print(memory.load_memory_variables({}))
  answer = agent_chain.run(json_data['text']['content'])
  json_send_message = {"msgtype": "text", "text": {"content": answer}}
  response = requests.post(DINGTALK_SEND_URL, headers={'Content-Type': 'application/json'}, json=json_send_message)
  print(response)
  return 'success'


if __name__ == '__main__':
  app.run(host='0.0.0.0', port=18888, debug=True) # 运转在有公网IP的服务器，一起开发18888端口

布置方式

我这里选用 Docker 布置。先创建一个简略 Dockerfile 用于构建镜像，Dockerfile 与上面的 webhook.py 和 requirements.txt 同一目录。

FROM python:3.9.17-slim-bullseye

WORKDIR /app

COPY requirements.txt requirements.txt
RUN pip3 install -r requirements.txt

COPY . .

CMD [ "python3", "webhook.py" ]

在 Terminal 中执行如下命令，完结镜像构建。

docker build -t ding-chatbot:1.0.4
docker images # 获取镜像ID cc07f3641130

我这里选用阿里云进行镜像管理。

docker tag cc07f3641130 registry.cn-hangzhou.aliyuncs.com/zwqh/ding-chatbot:1.0.4
docker push registry.cn-hangzhou.aliyuncs.com/zwqh/ding-chatbot:1.0.4

编写 docker-compose.yaml 布置脚本。

version: '3.9'
services:
  chatbot:
   image: registry.cn-hangzhou.aliyuncs.com/zwqh/ding-chatbot:1.0.4
   volumes:
    - ./app/.env:/app/.env
   ports:
    - 18888:18888
   networks:
    - xunlu

networks:
  xunlu:
   external: true

然后在 docker-compose.yaml 同级目录下创建 app 文件夹，并把 .env 文件放至该文件夹下。

在服务器上 docker-compose.yaml 目录下，经过以下命令完结布置。

docker-compose up -d

布置完结后，把公网可拜访的 URL 填写到钉钉开放渠道的消息接纳地址里。

运转相关日志

能够看不到 RealWeatherTool 东西主动被调用了。

> Entering new  chain...

Invoking: `RealWeatherTool` with `{}`


输入的城市信息或许有误或未供给城市信息请供给城市信息，例如：今天北京的气候怎么样？

> Finished chain.

演示效果

结束

本教程基于 LangChain + OpenAI 完结了气候实时查询及相关推理功能，而且支持镜像快速布置。

关注公众号【码森林】~

理解新范式，拥抱新时代，掌握新机会。

基于LangChain实现ChatGPT实时查询天气

LangChain

功能剖析（版别v1.0.4）

中心代码

依靠文件 requirements.txt

配置文件 .env

基于 BaseTool 完结自定义东西

程序入口 webhook.py

布置方式

运转相关日志

演示效果

结束

作者信息

推广

基于LangChain实现ChatGPT实时查询天气

LangChain

功能剖析（版别v1.0.4）

中心代码

依靠文件 requirements.txt

配置文件 .env

基于 BaseTool 完结自定义东西

程序入口 webhook.py

布置方式

运转相关日志

演示效果

结束

相关文章

TypeScript 命名空间、模块、声明合并 ｜ 青训营笔记

【游戏行业部】反外挂技术的革新：如何有效应对 FPS 外挂的威胁

ControlNet原理解析 | 读论文

大模型参数高效微调技术原理综述（五）-LoRA、AdaLoRA、QLoRA

作者信息

推广

TypeScript 命名空间、模块、声明合并｜青训营笔记