地理空间功能调用与ChatGPT

在这个例子中，大部分代码都是直接基于这本食谱的，并经过适应以作为一个用于执行基本地理空间函数的Python终端助手来进行说明。

使用ChatGPT进行函数调用使我们能够编写自定义软件和方法，这些方法可以由一个大型语言模型根据用户的自然语言输入来确定适当的函数和所需的参数。

所以将基本的想法应用于本教程 - 我们想要构建一个具有修复地理数据框架几何的能力以及执行一些简单地理空间算法（通过距离进行缓冲，并返回地理数据框架的边界框）的地理空间聊天机器人。现在这很简单，但纯粹是为了演示功能调用的概念 - 试想一下通过自然语言流程调用一系列执行复杂算法的地理空间API！

克隆代码库到这里

基本要求

现在这里的基本要求是你需要一个OpenAI开发者密钥 - 这里有一篇很棒的文章向你展示如何进行设置，我还有这个项目的完整代码，你可以克隆并使用 pip install -r requirements.txt 安装所需的所有依赖项。

您需要更新geo_chat.py文件，并将openai.api_key = "YOUR_API_KEY"替换为您自己的OpenAI密钥。

让我们来看一下我们的空间工具...

在我们的spatial_functions.py文件中，我们有执行我们的有趣地理空间工作的函数。

check_geom — 用于在检查地理数据帧的几何图形中评估各个几何图形的有效性。
check_geodataframe_geom（检查_geodataframe_geom）是repair_geodataframe_geom中用于评估geodataframe中是否存在任何无效几何的函数。
repair_geodataframe_geom - 修复地理数据框中的任何问题。
buffer_gdf- 正如其名称所示，该函数将geodataframe按指定距离进行缓冲（为了简化起见，我们不进行任何复杂的单位转换）。
bounding_box_of_gdf - 为 geodataframe 生成一个地理边界框

如果我们运行此文件，我们可以看到它针对一个硬编码的地理数据帧执行逻辑，并且我们修复它，缓冲它，并返回边界框几何 —— 您可以根据您的需求进行增加补充，但是我会保持本文中的内容非常简单。

from shapely.geometry import Polygon, LineString
import geopandas as gpd

# Define the initial set of invalid geometries as a GeoDataFrame
invalid_geometries = gpd.GeoSeries([
    Polygon([(0, 0), (0, 2), (1, 1), (2, 2), (2, 0), (1, 1), (0, 0)]),
    Polygon([(0, 2), (0, 1), (2, 0), (0, 0), (0, 2)]),
    LineString([(0, 0), (1, 1), (1, 0)]),
], crs='EPSG:3857')
invalid_polygons_gdf = gpd.GeoDataFrame(geometry=invalid_geometries)

# Function to check geometry validity
def check_geom(geom):
    return geom.is_valid

# Function to check all geometries in a GeoDataFrame
def check_geodataframe_geom(geodataframe: gpd.GeoDataFrame) -> bool:
    valid_check = geodataframe.geometry.apply(check_geom)
    return valid_check.all()

# Function to repair geometries in a GeoDataFrame
def repair_geodataframe_geom(geodataframe: gpd.GeoDataFrame)-> dict:
    if not check_geodataframe_geom(geodataframe):
        print('Invalid geometries found, repairing...')
        geodataframe = geodataframe.make_valid()
    return {"repaired": True, "gdf": geodataframe}

# Function to buffer all geometries in a GeoDataFrame
def buffer_gdf(geodataframe: gpd.GeoDataFrame, distance: float) -> gpd.GeoDataFrame:
    print(f"Buffering geometries by {distance}...")
    # Check type of distance
    if not isinstance(distance, (int, float)):
        raise TypeError("Distance must be a number")
    
    # Applying a buffer to each geometry in the GeoDataFrame
    buffered_gdf = geodataframe.copy()
    buffered_gdf['geometry'] = buffered_gdf.geometry.buffer(distance)
    return {"message": "Geometries buffered successfully", "gdf": buffered_gdf}

# Function to get the bounding box of all geometries in a GeoDataFrame
def bounding_box_of_gdf(geodataframe: gpd.GeoDataFrame):
    # get bounding box of geodataframe 
    bbox = geodataframe.total_bounds
    return {"message": "Bounding box obtained successfully", "bbox": bbox}

# Main execution block
if __name__ == '__main__':
    print("Checking and repairing geometries...")
    repaired_polygons_gdf = repair_geodataframe_geom(invalid_polygons_gdf)
    
    all_geometries_valid = check_geodataframe_geom(repaired_polygons_gdf)
    print(f"All geometries valid: {all_geometries_valid}")
    
    # Example of buffering the geometries
    buffered_polygons_gdf = buffer_gdf(repaired_polygons_gdf, 0.1)
    
    # Getting the bounding box of the geometries
    bbox = bounding_box_of_gdf(buffered_polygons_gdf)
    print(f"Bounding box: {bbox}")

但是现在我希望用户能够轻松使用这些功能- 只需简单地指示聊天机器人执行这些操作。

构建我们的聊天机器人

这就是我们制作一个简单但能够调用地理空间功能的聊天机器人所需的一切 - 现在最困难的部分是对话流程/处理。我们可以花很多时间来优化对话的流程，但在这里我们只做最低限度的工作。

首先，让我们导入和声明我们需要使用的所有内容，确保使用gpt-3.5-turbo-0613来完成本教程。

import json
import openai
import requests
from tenacity import retry, wait_random_exponential, stop_after_attempt
from termcolor import colored

# Import our spatial functions
from spatial_functions import *

# Just Hard coding our example data for this demo from our spatial_functions.py
DEMO_GEODATAFRAME = invalid_polygons_gdf

# GPT Variables
GPT_MODEL = "gpt-3.5-turbo-0613"
openai.api_key = "YOUR_API_KEY"

现在让我们声明一些函数，以使对话处理变得更容易一些：

聊天完成请求

一个Python函数，使用@retry装饰器自动重试对OpenAI Chat API的请求，在尝试3次后停止重试。这个函数主要处理我们与聊天API的通信，并处理我们发送和接收对话数据的方式。

@retry(wait=wait_random_exponential(multiplier=1, max=40), stop=stop_after_attempt(3))
def chat_completion_request(messages, tools=None, tool_choice=None, model=GPT_MODEL):
    headers = {
        "Content-Type": "application/json",
        "Authorization": "Bearer " + openai.api_key,
    }
    json_data = {"model": model, "messages": messages}
    if tools is not None:
        json_data.update({"tools": tools})
    if tool_choice is not None:
        json_data.update({"tool_choice": tool_choice})
    try:
        response = requests.post(
            "https://api.openai.com/v1/chat/completions",
            headers=headers,
            json=json_data,
        )
        return response
    except Exception as e:
        print("Unable to generate ChatCompletion response")
        print(f"Exception: {e}")
        return e

漂亮打印对话

这个函数的主要目的是让我们的聊天对话在终端中看起来好看 — 它根据消息被分配的角色（基本上是谁说了什么？你，聊天机器人还是一个通用系统消息）来为消息着色，使对话对我们用户来说更合乎逻辑。

def pretty_print_conversation(messages):
    role_to_color = {
        "system": "red",
        "user": "green",
        "assistant": "blue",
    }
    
    for message in messages:
        if message["role"] == "system":
            print(colored(f"system: {message['content']}\n", role_to_color[message["role"]]))
        elif message["role"] == "user":
            print(colored(f"user: {message['content']}\n", role_to_color[message["role"]]))
        elif message["role"] == "assistant" and message.get("function_call"):
            print(colored(f"assistant: {message['function_call']}\n", role_to_color[message["role"]]))
        elif message["role"] == "assistant" and not message.get("function_call"):
            print(colored(f"assistant: {message['content']}\n", role_to_color[message["role"]]))

好的，很棒，但我们如何让Chat GPT理解我们的功能呢？

我们需要指导ChatGPT关于它执行地理空间功能所需的要素。根据与用户的对话，有一个很棒的工具叫做工具，如果我们设定好了，就基本上是指导ChatGPT执行我们希望用户执行的功能，它们存在的上下文以及我们希望大型语言模型从用户的文本中得出的任何参数。

tools = [
    # Our Repair Function
    {
        "type": "function",
        "function": {
            "name": "repair_geodataframe_geom",
            "description": "Repair invalid geometries in a GeoDataFrame",
            "parameters": {},
        }
    },
    # Our Bounding Box Function
    {
        "type": "function",
        "function": {
            "name": "bounding_box_of_gdf",
            "description": "Get the geospatial bounding box of a GeoDataFrame",
            "parameters": {},
        }
    },
    # Our Buffer Function
    {
        "type": "function",
        "function": {
            "name": "buffer_gdf",
            "description": "Buffer a GeoDataFrame by a specified distance",
            "parameters": {
                "type": "object",
                "properties": {
                    "distance": {
                        "type": "string",
                        "description": "A specific distance as a number that will be used to buffer the GeoDataFrame",
                    },
                },
                "required": ["distance"],
            },
        }
    },
]

好的，所以根据我们声明的工具的顺序 —— 它们显然都是函数对吧？我们必须指定函数的名称，这样如果模型识别出函数在用户的消息中的上下文，它就会知道这是我需要调用的函数名称，因此我可以查找它需要运行什么。

描述提供背景信息:

现在，描述本质上为模型提供了用户可能会问的上下文，所以如果我们看一下 repair_geodataframe_geom，我们会发现描述是“修复GeoDataFrame中的无效几何图形”。因此，当用户问“你能修复我的geodataframe吗？”时，它将查看给定的工具，并且会说“好的，我有一个与该上下文匹配的工具”。

参数但如何

现在为了简单起见，在 repair_geodataframe_geom 和 bounding_box_of_gdf 中，我们没有从用户那里获取参数（如果我们愿意，也可以获取，但对于这个演示而言并不需要），但是为了说明，我们的缓冲函数需要一个距离，它需要从用户那里知道需要缓冲多少半径，因此我们为我们的工具设置了参数：

"parameters": {
    "type": "object",
    "properties": {
        "distance": {
            "type": "string",
            "description": "A specific distance as a number that will be used to buffer the GeoDataFrame",
        },
    },
    "required": ["distance"],
},

而这样做的作用是我们可以看到，描述再次提供上下文，因此如果用户说我想将我的地理数据帧缓冲20的距离，模型将理解数字20属于距离参数，它会提取20并将其分配为距离--距离是必需的，显然我们不希望该函数在没有获取所需一切的情况下执行，因此我们将确保我们的指令有足够的细节来提供ChatGPT所需的内容。

构建对话

好的，正如我所提到的，对话流和处理消息可能可以作为一个独立的系列文章，所以我只会告诉你相信我在这里设置代码流程的方式-它非常简单，我们有很多方法可以处理对话，但我们只会关注如何让模型执行我们的函数的具体逻辑。

所以这是我们的对话（大的代码块将讨论关键事项）：

if __name__ == "__main__":

  messages = []
  messages.append({"role": "system", "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."})

  while True:
      user_input = input("You: ")
      if user_input.lower() == "exit":
          break

      messages.append({"role": "user", "content": user_input})

      chat_response = chat_completion_request(messages, tools=tools)

      if chat_response.status_code == 200:
          response_data = chat_response.json()
          assistant_message_content = response_data["choices"][0].get("message", {}).get("content")

          # Ensure assistant_message_content is not None before appending
          if assistant_message_content:
              messages.append({"role": "assistant", "content": assistant_message_content})
          else:
              messages.append({"role": "assistant", "content": "Of course I can!"})

          tool_calls = response_data["choices"][0]["message"].get("tool_calls", [])
          
          for tool_call in tool_calls:
              function_call = tool_call["function"]
              tool_name = function_call["name"]

              if tool_name == "repair_geodataframe_geom":
                  repair_result = repair_geodataframe_geom(DEMO_GEODATAFRAME)
                  tool_message = "GeoDataFrame repair completed."

              elif tool_name == "bounding_box_of_gdf":
                  response = bounding_box_of_gdf(DEMO_GEODATAFRAME)
                  message = response["message"]
                  bbox = response["bbox"]
                  tool_message = f"{message} {bbox}"

              elif tool_name == "buffer_gdf":
                  function_arguments = json.loads(function_call["arguments"])
                  distance = function_arguments["distance"]
                  response = buffer_gdf(DEMO_GEODATAFRAME, int(distance))
                  DEMO_GEODATAFRAME = response["gdf"]
                  tool_message = f"The GeoDataFrame has been buffered by {distance}."
              
              else:
                  tool_message = f"Tool {tool_name} not recognized or not implemented."
              messages.append({"role": "assistant", "content": tool_message})

          # Print the conversation with the assistant
          pretty_print_conversation(messages)

      else:
          print(f"Failed to get a response from the chat service. Status Code: {chat_response.status_code}")
          try:
              error_details = chat_response.json()
              print("Response error details:", error_details.get("error", {}).get("message"))
          except Exception as e:
              print(f"Error parsing the error response: {e}")

  print("\nConversation ended.")

请注意，在我们开始与ChatGPT的对话之前，我们将此消息附加到我们的对话中。messages.append({"role": "system", "content": "在将值插入函数之前，请不要做任何假设。如果用户请求不明确，请征求澄清。"}) — 这是ChatGPT将处理的第一个提示。

这就是我们所称之为系统消息的内容，它会指示ChatGPT确保用户提供了有关功能参数的所有细节，比如如果用户说“我想要缓冲我的地理数据框”，我们的模型就会知道它需要调用buffer_gdf函数，但是它缺少距离参数，因此ChatGPT需要与用户澄清这个问题，并且我们会在演示中展示出来。所以请记住这一点。

对话以你为中心开始。

while True:
  user_input = input("You: ")
  if user_input.lower() == "exit":
      break

  messages.append({"role": "user", "content": user_input})

所以我们的应用程序是一个终端聊天机器人，基本上我们等待用户指定一条消息 - 当他们写一条消息时，我们初始化我们的ChatGPT模型 - 我们将聊天的消息和我们声明的工具传递给模型，这样模型就知道了：“我正在进行一次对话，这里是我需要注意并监控对话的工具”。

chat_response = chat_completion_request(messages, tools=tools)

这个下一部分处理从ChatGPT模型接收到的响应，并规定了对话需要如何进行：

if chat_response.status_code == 200:
    response_data = chat_response.json()
    assistant_message_content = response_data["choices"][0].get("message", {}).get("content")

    # Ensure assistant_message_content is not None before appending
    if assistant_message_content:
        messages.append({"role": "assistant", "content": assistant_message_content})
    else:
        messages.append({"role": "assistant", "content": "Of course I can!"})

    tool_calls = response_data["choices"][0]["message"].get("tool_calls", [])
...

else:
  print(f"Failed to get a response from the chat service. Status Code: {chat_response.status_code}")
  try:
      error_details = chat_response.json()
      print("Response error details:", error_details.get("error", {}).get("message"))
  except Exception as e:
      print(f"Error parsing the error response: {e}")

如果反应良好，我们从反应中获取聊天数据，并从ChatGPT模型中识别消息，然后我们想检查模型是否已识别出我们的工具之一是否被调用。如果反应出现任何问题，我们只需进行一些错误处理。

所以，如果我说“你好”，响应中没有任何工具调用，我将从ChatGPT收到一条普通消息。

如果响应中有工具调用，则意味着ChatGPT模型已经确定用户指定的上下文与关于我们工具的指示相吻合。

这段代码遍历响应并且如果有工具调用，则识别出ChatGPT模型识别出来的工具，并相应地执行它。

for tool_call in tool_calls:
    function_call = tool_call["function"]
    tool_name = function_call["name"]

    if tool_name == "repair_geodataframe_geom":
        repair_result = repair_geodataframe_geom(DEMO_GEODATAFRAME)
        tool_message = "GeoDataFrame repair completed."

    elif tool_name == "bounding_box_of_gdf":
        response = bounding_box_of_gdf(DEMO_GEODATAFRAME)
        message = response["message"]
        bbox = response["bbox"]
        tool_message = f"{message} {bbox}"

    elif tool_name == "buffer_gdf":
        function_arguments = json.loads(function_call["arguments"])
        distance = function_arguments["distance"]
        response = buffer_gdf(DEMO_GEODATAFRAME, int(distance))
        DEMO_GEODATAFRAME = response["gdf"]
        tool_message = f"The GeoDataFrame has been buffered by {distance}."
    
    else:
        tool_message = f"Tool {tool_name} not recognized or not implemented."
    
messages.append({"role": "assistant", "content": tool_message})

那么让我们请模型缓冲我们的地理数据框架 - 默认情况下，我们的代码知道只需使用我们的 DEMO_GEODATAFRAME。

注意，初始时我们的请求不明确（我们忽略了具体的距离），ChatGPT需要我们澄清一些事情 - 因此，在这种情况下，我们必须指定距离为20，然后ChatGPT使用所需的信息执行函数，在DEMO_GEODATAFRAME上执行一个20的缓冲操作。

就像魔术一样，对吧？让我们来看看与我们的新地理聊天机器人进行的完整对话：

总之，

AI工具的进化无疑为持续创新打开了大门 — 这只是Open AI团队开发的一项非常强大功能的小示范，而且具有广泛的潜在应用。我目前正在利用这种功能进行一个更大的项目，敬请关注。

我希望像这样的文章能够激发您利用这些令人惊叹的工具集来拓展您的工作，并继续为地理空间世界开发创新解决方案。

如果您有任何问题，请随时联系我们，再次感谢大家！

克隆存储库在这里

社交媒体

Fosstodon: https://fosstodon.org/@GeoScrubJohnL Fosstodon：https://fosstodon.org/@GeoScrubJohnL
Twitter/X：https://twitter.com/GeoScrubJohnL
领英：www.linkedin.com/in/john-lister-854422209

请查看我的新播客节目GeoData。