利用 ChatGPT 的函数调用功能，我们可以编写自定义软件和方法，让大型语言模型根据用户的自然语言输入来决定适当的函数和所需的参数。

我们要构建一个地理空间聊天机器人，它能修复地理数据框的几何图形，并执行一些简单的地理空间算法（按距离缓冲，并返回地理数据框的边界框）。

基本要求

这里的基本要求是，你必须拥有 OpenAI Developer 密钥，你需要更新 geo_chat.py 文件，并将 openai.api_key = "YOUR_API_KEY" 替换为你自己的 OpenAI 密钥。

工具

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

• check_geom - 在 check_geodataframe_geom 中使用，以评估单个几何图形的有效性。
• check_geodataframe_geom - 在 repair_geodataframe_geom 中使用 check_geodatafrane_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"

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

chat_completion_requests

这是一个 Python 函数，使用 @retry 装饰器自动重试向 OpenAI 聊天应用程序接口发出的请求，它会在尝试 3 次后停止重试--该函数主要处理我们与聊天应用程序接口的通信方式，以及我们发送和接收对话数据的方式。

@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

pretty_print_conversation

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

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 它需要什么，以便根据与用户的对话执行地理空间功能。有一个很有趣的东西叫做tools，如果我们设置了它，基本上就能指示 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 和 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.")

请注意，我们将此消息附加到我们的对话中，然后再进行任何操作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.”})- 当我们与 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 模型中识别消息，并检查该模型是否已识别我们的工具是否被调用。如果回复有任何问题，我们就会进行错误处理。

因此，如果我说 "hello"，响应中就不会有工具调用，所以我会从 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 的缓冲。

让我们来看看新地理空间聊天机器人的完整对话：

结论

人工智能工具的发展无疑为不断创新打开了大门--这只是开放式人工智能团队开发的一项极其强大的功能的一个小演示，它具有广泛的潜在应用。