当前位置：首页 > news >正文

如何使用 Google Gemini API 和 Python 从航行情报通告 (NOTAM) 中提取结构化空域信息

news 来源：原创 2025/5/16 7:41:27

首先给大家准备一些gemini的api-key 方便大家实战gemini。
AIzaSyDFKzqbwbM78iflpIlgj81cSAaEOWjbzJs
AIzaSyBq_qvVZqp8InnNrkuAPCpJBC7EWiAnAVc
AIzaSyAxc7owaBrs9QbK9fzGplV6enyUt6T2aOg
AIzaSyCy4kznxmWi1OkLkaffZ6mL-RpVaUoeXEs
AIzaSyCj_dvlvKdf87b2zdyycFhcdHRuOigzSZQ
AIzaSyDhguhSVoZ_5EIbN2p63hDWKxNFFZBL_2s
AIzaSyBPHGM30gG4YHrTYI4yUEXYCJkTRjgYT80
AIzaSyDAJ8zQAl8nwA_JjDEf7WmAkqP8k9f5r-c

接下来，这是一个关于如何使用 Google Gemini API 和 Python 从航行情报通告 (NOTAM) 中提取结构化空域信息的教程。本教程将基于我们之前的讨论，包括定义数据结构、构建提示、进行 API 调用以及处理可能出现的 schema 问题。

教程：使用 Gemini API 和 Python 从 NOTAM 提取结构化空域信息

目标：

学习如何设置和使用 Google Generative AI Python SDK。
使用 Pydantic 定义期望的输出数据结构。
构建有效的提示 (prompt) 以指导 Gemini API 进行信息提取。
调用 Gemini API 并获取结构化的 JSON 响应。
处理 API 响应，包括使用 response_schema (如果支持复杂类型) 和手动 JSON 解析作为备选方案。

前提条件：

Python 环境: 确保您已安装 Python (建议 3.9 或更高版本)。
Google AI API 密钥:
- 您需要一个 Google Cloud 项目并启用 Vertex AI API，或者通过 Google AI Studio 获取 API 密钥。
- 获取 API 密钥后，请妥善保管。

安装必要的库:

pip install google-generativeai pydantic

步骤 1：设置项目和导入库

首先，创建一个 Python 文件 (例如 notam_parser.py) 并导入所需的库。

import google.generativeai as genai
from pydantic import BaseModel, TypeAdapter
import json
import os # 用于安全处理 API 密钥 (推荐)# 用于清晰的类型提示 (Python 3.9+ 可以直接使用 list, dict)
# from typing import List, Dict

步骤 2：配置 API 密钥

安全地配置您的 API 密钥。建议使用环境变量。

# 推荐方式：从环境变量读取 API 密钥
# 在您的终端设置：export GOOGLE_API_KEY="YOUR_API_KEY"
# 或者，如果您在类似 Jupyter Notebook 的环境中，可以直接设置，但不推荐用于生产代码
# API_KEY = "YOUR_API_KEY" # 直接替换 YOUR_API_KEY，注意安全风险try:# 尝试从环境变量获取 API 密钥API_KEY = os.environ["GOOGLE_API_KEY"]
except KeyError:print("错误：GOOGLE_API_KEY 环境变量未设置。")print("请设置该变量或在代码中直接提供 API_KEY (注意安全)。")# 为防止代码完全中断，您可以临时在这里设置，但强烈建议使用环境变量API_KEY = "YOUR_API_KEY_PLACEHOLDER" # <--- 临时的，请务必替换或通过环境变量设置if API_KEY == "YOUR_API_KEY_PLACEHOLDER":raise ValueError("请将 YOUR_API_KEY_PLACEHOLDER 替换为您的有效 API 密钥或设置 GOOGLE_API_KEY 环境变量。")genai.configure(api_key=API_KEY)

请务必将 "YOUR_API_KEY_PLACEHOLDER" 替换为您的真实 API 密钥，或通过设置 GOOGLE_API_KEY 环境变量来提供。

步骤 3：定义输出数据结构 (Pydantic 模型)

我们将定义一个 Pydantic 模型来描述从 NOTAM 中提取的每个字段的结构。我们期望的最终输出是一个包含多个“区域”信息的列表，每个“区域”信息又包含多个提取字段。

class ExtractedField(BaseModel):start: intend: intlabel: str  # 例如: "通告号", "机场名称" 等中文标签text: str# 期望的整体输出结构是: list[list[ExtractedField]]
# 即一个列表，其中每个元素是代表一个空域区域信息的字段列表

步骤 4：准备输入数据 (NOTAM 文本和指令)

这是我们要处理的 NOTAM 文本和给模型的指令。

instruction_text = "检测到空域相关关键词。请参照参考案例的结构和字段标签，从航行情报通告中提取所有相关的空域信息。"
notam_to_process = """ZCZC BUI2865 251937 GG ZBAAUIXX251937 ZBAAOIXXUUUU(K1372/25 NOTAMNQ)UUWV/QRTCA/IV/BO/W/000/030/5549N03844E003A)UUWV B)2501290900 C)2502031600 D)JAN 29-31 FEB 01 03 0900-1600E) AIRSPACE CLSD WI AREA:554930N0384015E-554916N0383931E-554903N0384027E-554902N0384104E-554905N0384411E-554909N0384845E-554940N0384840E-554930N0384512E-554925N0384106E-554930N0384015EF)SFC G)640M AMSL)NNNN"""

步骤 5：构建提示 (Prompt)

一个好的提示对于引导模型输出正确的结构化数据至关重要。

# 详细的提示，指导模型进行提取并遵循特定的JSON结构
# 注意：这里的 {notam_to_process} 将会被实际的NOTAM文本替换
prompt_template = f"""指令: {instruction_text}待处理的航行情报通告 (NOTAM) 文本:
---
{notam_to_process}
---请从上述 NOTAM 文本中提取所有相关的空域信息。
您的输出必须是一个严格的 JSON 格式字符串。
此 JSON 字符串应代表一个JSON数组，该数组的每个元素本身也是一个JSON对象列表 (即 `list[list[object]]` 结构)。
每个JSON对象代表一个提取出的字段，并且必须包含以下键值对:
- "start": (整数) 提取文本在原始 NOTAM 字符串中的起始字符索引 (0-based)。
- "end": (整数) 提取文本在原始 NOTAM 字符串中的结束字符索引 (不包含结束字符本身)。
- "label": (字符串) 字段的中文标签 (例如："通告号", "机场名称", "多边形坐标序列" 等，请参考通用NOTAM字段定义和标签)。
- "text": (字符串) 为该字段提取的实际文本。例如，期望的 JSON 结构是：`[[{{"start": 0, "end": 5, "label": "示例标签", "text": "示例文本"}}]]`
如果NOTAM中包含多个空域区域的描述，请为每个区域生成一个内部列表。对于给定的NOTAM，它描述了一个主要区域。
"""

步骤 6：进行 API 调用和处理响应

我们将尝试两种方法：

方法 A (推荐，如果SDK支持您的schema): 使用 response_schema 来让 SDK 自动解析和验证。
方法 B (备选方案): 如果方法 A 对于 list[list[ExtractedField]] 这样的复杂嵌套 schema 失败 (出现 Unsupported schema type 错误)，我们将移除 response_schema 并手动解析 JSON。

# 初始化 Gemini 模型客户端
# 您最初使用的是 client.models.generate_content，这里我们使用更通用的方式获取模型实例
# 但会保持与您原始代码相似的参数传递方式 (使用 'config' 而非 'generation_config')# 使用您在原始代码中指定的模型
model_name = "models/gemini-2.5-flash-preview-04-17"
# 您也可以尝试其他模型，例如 "gemini-1.5-flash-latest"
# model_name = "gemini-1.5-flash-latest"model = genai.GenerativeModel(model_name)print(f"--- 方法 A: 尝试使用 response_schema ---")
try:# 注意：client.models.generate_content() 是您之前用的方式。# 如果直接使用 model.generate_content()，参数名可能是 generation_config。# 为了与您之前的代码和错误保持一致，我们先模拟 client.models.generate_content() 的参数风格# 但实际上 genai.GenerativeModel(model_name).generate_content() 是更推荐的用法。# 为简单起见，这里直接用 model.generate_content 并使用正确的 generation_config。# 如果您坚持使用 client.models.generate_content 且它存在并使用 'config'，请相应调整。# 经过之前的讨论，我们知道 client.models.generate_content(config=...) 这种方式# 对于 list[list[ExtractedField]] 可能会报 "Unsupported schema type"# 因此，方法A的演示将直接跳到结论，并引导至方法B。# 或者，我们可以先尝试，如果您的SDK版本支持，它就能工作。print("尝试使用 config 和 response_schema...")# 这里我们用 model.generate_content，它使用 generation_config# 如果您想严格模拟 client.models.generate_content，您需要确保该方法存在且可用。# genai.Client 本身没有 .models.generate_content()。# 正确的方式是 client.get_generative_model(model_name).generate_content()# 或者 genai.GenerativeModel(model_name).generate_content()# 我们将使用 genai.GenerativeModel().generate_content()，它使用 'generation_config'# 并且，如果 'response_schema': list[list[ExtractedField]] 仍然不受支持，# 我们将看到同样的错误，然后转到方法B。response_attempt_A = model.generate_content(prompt_template, # 使用上面定义的完整 promptgeneration_config=genai.types.GenerationConfig( # 显式使用 GenerationConfig 对象response_mime_type="application/json",# response_schema=list[list[ExtractedField]] # 这个是导致 "Unsupported schema type" 的原因# 我们知道这个可能会失败，所以会注释掉并演示手动解析))print("\n原始JSON响应 (方法 A - 无 schema 强制，因已知问题):")print(response_attempt_A.text)print("\n方法 A 结论：对于 list[list[PydanticModel]] 这样的复杂嵌套 schema，")print("response_schema 可能不被直接支持，导致 'Unsupported schema type' 错误。")print("我们将主要依赖方法 B (手动解析)。")except Exception as e:print(f"\n方法 A 执行时发生错误: {e}")print("这可能是因为 response_schema 不支持 list[list[ExtractedField]] 结构。")print("将继续使用方法 B。")print(f"\n\n--- 方法 B: 手动解析 JSON (更可靠的方案) ---")
try:# 对于手动解析，我们不设置 response_schema 或 response_mime_type，# 而是依赖提示清晰地要求JSON输出。response_attempt_B = model.generate_content(prompt_template) # 无 generation_config 中的 schemaprint("\n原始响应文本 (方法 B):")if response_attempt_B.text:print(response_attempt_B.text)# 手动解析 JSONraw_data = json.loads(response_attempt_B.text)# 使用 Pydantic TypeAdapter 验证和转换数据# 这是我们期望的最终数据结构expected_schema_adapter = TypeAdapter(list[list[ExtractedField]])extracted_data = expected_schema_adapter.validate_python(raw_data)print("\n已解析和验证的 Pydantic 对象 (方法 B):")if not extracted_data:print("解析后的数据为空列表。")for i, area_fields_list in enumerate(extracted_data):print(f"区域 {i+1}:")if not area_fields_list:print(f"  区域 {i+1} 的字段列表为空。")for field_obj in area_fields_list:print(f"  标签: {field_obj.label}, 文本: '{field_obj.text}', 起始: {field_obj.start}, 结束: {field_obj.end}")else:print("\n模型返回的响应文本为空 (方法 B)。")# 打印其他响应信息以供调试if hasattr(response_attempt_B, 'prompt_feedback') and response_attempt_B.prompt_feedback:print(f"\n提示反馈 (方法 B): {response_attempt_B.prompt_feedback}")if response_attempt_B.candidates:for candidate_idx, candidate in enumerate(response_attempt_B.candidates):finish_reason_name = candidate.finish_reason.name if hasattr(candidate.finish_reason, 'name') else 'N/A'print(f"\n候选 {candidate_idx} (方法 B):")print(f"  完成原因: {candidate.finish_reason} ({finish_reason_name})")if hasattr(candidate, 'finish_message') and candidate.finish_message:print(f"  完成消息: {candidate.finish_message}")if hasattr(candidate, 'safety_ratings') and candidate.safety_ratings:for rating in candidate.safety_ratings:category_name = rating.category.name if hasattr(rating.category, 'name') else rating.categoryprobability_name = rating.probability.name if hasattr(rating.probability, 'name') else rating.probabilityprint(f"  安全评级: 类别={category_name}, 概率={probability_name}")else:print("\n响应中未找到候选内容 (方法 B)。")except json.JSONDecodeError as e:print(f"\nJSON 解码错误 (方法 B): {e}")print("模型返回的文本可能不是有效的 JSON。请检查原始响应文本。")
except Exception as e: # 例如 Pydantic ValidationError 或其他API错误print(f"\n在调用 API 或处理响应时发生错误 (方法 B): {e}")import tracebacktraceback.print_exc()

教程解释：

API 密钥和库设置: 标准的初始步骤。
Pydantic 模型 (ExtractedField): 定义了我们希望从 NOTAM 中为每个信息片段提取的具体字段。这有助于后续的数据验证和使用。
输入数据: 我们使用了之前讨论中的 NOTAM 文本和提取指令。
提示构建 (prompt_template):
- 清晰地告诉模型它的任务 (instruction_text)。
- 提供要处理的原始文本 (notam_to_process)。
- 极其重要: 明确指定输出必须是 JSON 字符串，并详细描述了期望的 JSON 结构 (list[list[object]]，其中每个对象包含 start, end, label, text)。给出一个小例子也有助于模型理解。
API 调用与响应处理:
- 模型初始化: genai.GenerativeModel(model_name) 是获取模型实例的推荐方式。
- 方法 A (注释掉的部分): 我们之前的讨论表明，直接使用 response_schema=list[list[ExtractedField]] 可能会因为 SDK 不支持这种深层嵌套的 schema 类型而失败。如果您的 SDK 版本或调用方式有所不同且支持此功能，您可以取消注释并尝试。
- 方法 B (手动解析):
  - 我们调用 model.generate_content(prompt_template) 时不在 generation_config 中指定 response_schema 或 response_mime_type。
  - 我们依赖提示来确保模型返回 JSON 字符串。
  - 获取 response.text。
  - 使用 json.loads() 将 JSON 字符串转换为 Python 字典/列表。
  - 使用 pydantic.TypeAdapter(list[list[ExtractedField]]).validate_python(raw_data) 来验证加载的数据是否符合我们定义的 ExtractedField 结构，并将其转换为 Pydantic 对象。这一步提供了类型安全和数据验证。
  - 最后，遍历提取的数据并打印。
- 调试信息: 打印 prompt_feedback 和 candidates 信息有助于了解 API 调用的详细情况，例如是否有内容被阻止或提前终止。

关键点和最佳实践：

清晰的提示: 对于结构化数据提取，提示的质量至关重要。明确描述期望的输出格式 (JSON)、结构 (字段名、嵌套层级) 和内容类型。
Pydantic: 使用 Pydantic 模型来定义数据结构可以使代码更健壮，易于维护，并提供数据验证。
response_schema 的局限性: 虽然 response_schema 是一个强大的功能，但对于非常复杂的或深层嵌套的列表结构，它可能存在局限性 (如 Unsupported schema type 错误)。
手动 JSON 解析作为备选: 如果 response_schema 不起作用，手动解析 response.text 并结合 Pydantic TypeAdapter进行验证是一个非常可靠的备选方案。
错误处理: 包含 try-except 块来捕获潜在的 API 错误、JSON 解析错误或 Pydantic 验证错误。
API 密钥安全: 始终优先使用环境变量来存储和访问 API 密钥。
迭代和测试: 构建提示和解析逻辑通常需要一些迭代。从简单的例子开始，逐步增加复杂性，并不断测试。

这个教程应该能帮助您开始使用 Gemini API 从文本中提取结构化数据。根据您的具体需求和遇到的情况，您可能需要调整提示或数据处理逻辑。

完整可运行代码

from google import genai
from pydantic import BaseModel
# from typing import List, Optional # Removed or not used for schema# 1. 定义新的 Pydantic 模型以匹配 NOTAM 提取的输出结构
class ExtractedField(BaseModel):start: intend: intlabel: strtext: strAPI_KEY = "AIzaSyAmf-glmlPjFbeXaWGXlFo8uVUE-4-7eZ4" # <--- 请在这里替换为您的有效 API 密钥
client = genai.Client(api_key=API_KEY)instruction_text = "检测到空域相关关键词。请参照参考案例的结构和字段标签，从航行情报通告中提取所有相关的空域信息。"
notam_text = """ZCZC BUI2865 251937 GG ZBAAUIXX251937 ZBAAOIXXUUUU(K1372/25 NOTAMNQ)UUWV/QRTCA/IV/BO/W/000/030/5549N03844E003A)UUWV B)2501290900 C)2502031600 D)JAN 29-31 FEB 01 03 0900-1600E) AIRSPACE CLSD WI AREA:554930N0384015E-554916N0383931E-554903N0384027E-554902N0384104E-554905N0384411E-554909N0384845E-554940N0384840E-554930N0384512E-554925N0384106E-554930N0384015EF)SFC G)640M AMSL)NNNN"""prompt = f"""指令: {instruction_text}待处理的航行情报通告 (NOTAM) 文本:
---
{notam_text}
---请从上述 NOTAM 文本中提取所有相关的空域信息。
输出必须是一个 JSON 数组。此数组中的每个元素本身也应该是一个对象列表。
每个对象代表一个提取出的字段，并且必须包含以下属性:
- "start": 提取文本在原始 NOTAM 字符串中的起始字符索引 (0-based)。
- "end": 提取文本在原始 NOTAM 字符串中的结束字符索引 (不包含结束字符本身)。
- "label": 字段的中文标签 (例如："通告号", "机场名称", "多边形坐标序列" 等，需符合指令中提及的参考案例和字段标签规则)。
- "text": 为该字段提取的实际文本。当检测到“空域相关关键词”时，请参照标准的 NOTAM 解析实践来识别字段及其对应的中文标签。
请确保输出严格遵循指定的 JSON 结构：一个包含多个字段对象列表的列表 (list of lists of field objects)。
例如，如果一个 NOTAM 描述了多个独立的空域区域，则每个区域提取的信息（包括为该区域上下文重复的通用 NOTAM 细节以及该区域的特定细节）应构成一个内部的字段对象列表。如果只描述了一个区域，则外部列表将包含一个内部的字段列表。
"""try:response = client.models.generate_content(model="models/gemini-2.5-flash-preview-04-17",contents=prompt,config={"response_mime_type": "application/json",# Use built-in list for the schema definition"response_schema": list[list[ExtractedField]],},)print("原始 JSON 响应:")print(response.text)if hasattr(response, 'parsed') and response.parsed:# Adjust type hint for extracted_data if needed, though it's for readabilityextracted_data: list[list[ExtractedField]] = response.parsedprint("\n已解析的 Pydantic 对象:")if not extracted_data:print("解析后的数据为空列表。")for i, area_fields in enumerate(extracted_data):print(f"区域 {i+1}:")if not area_fields:print(f"  区域 {i+1} 的字段列表为空。")for field_obj in area_fields:print(f"  标签: {field_obj.label}, 文本: '{field_obj.text}', 起始: {field_obj.start}, 结束: {field_obj.end}")elif response.text is None or response.text.strip() == "":print("\n模型返回的响应文本为空。")else:print("\n未能从 response.parsed 获取已解析的对象，或者解析结果为空。")if hasattr(response, 'prompt_feedback') and response.prompt_feedback:print(f"提示反馈: {response.prompt_feedback}")if response.candidates:for candidate_idx, candidate in enumerate(response.candidates):print(f"候选 {candidate_idx} 完成原因: {candidate.finish_reason} ({candidate.finish_reason.name if hasattr(candidate.finish_reason, 'name') else 'N/A'})")if hasattr(candidate, 'finish_message') and candidate.finish_message:print(f"候选 {candidate_idx} 完成消息: {candidate.finish_message}")if hasattr(candidate, 'safety_ratings') and candidate.safety_ratings:for rating in candidate.safety_ratings:print(f"候选 {candidate_idx} 安全评级: 类别={rating.category.name if hasattr(rating.category, 'name') else rating.category}, 概率={rating.probability.name if hasattr(rating.probability, 'name') else rating.probability}")else:print("响应中未找到候选内容。")except Exception as e:print(f"\n在调用 API 或处理响应时发生错误: {e}")import tracebacktraceback.print_exc()

输出

原始 JSON 响应:
[[{"start": 54,"end": 68,"label": "通告号","text": "K1372/25 NOTAMN"},{"start": 69,"end": 113,"label": "Q码","text": "Q)UUWV/QRTCA/IV/BO/W/000/030/5549N03844E003"},{"start": 114,"end": 120,"label": "影响范围","text": "A)UUWV"},{"start": 121,"end": 133,"label": "起始时间","text": "B)2501290900"},{"start": 134,"end": 146,"label": "结束时间","text": "C)2502031600"},{"start": 147,"end": 177,"label": "计划时间","text": "D)JAN 29-31 FEB 01 03 0900-1600"},{"start": 178,"end": 336,"label": "空域描述","text": "E) AIRSPACE CLSD WI AREA:\n554930N0384015E-554916N0383931E-\n554903N0384027E-554902N0384104E-\n554905N0384411E-554909N0384845E-\n554940N0384840E-554930N0384512E-\n554925N0384106E-554930N0384015E"},{"start": 204,"end": 336,"label": "多边形坐标序列","text": "554930N0384015E-554916N0383931E-\n554903N0384027E-554902N0384104E-\n554905N0384411E-554909N0384845E-\n554940N0384840E-554930N0384512E-\n554925N0384106E-554930N0384015E"},{"start": 336,"end": 341,"label": "限制的下边界","text": "F)SFC"},{"start": 342,"end": 353,"label": "限制的上边界","text": "G)640M AMSL"}]
]已解析的 Pydantic 对象:
区域 1:标签: 通告号, 文本: 'K1372/25 NOTAMN', 起始: 54, 结束: 68标签: Q码, 文本: 'Q)UUWV/QRTCA/IV/BO/W/000/030/5549N03844E003', 起始: 69, 结束: 113标签: 影响范围, 文本: 'A)UUWV', 起始: 114, 结束: 120标签: 起始时间, 文本: 'B)2501290900', 起始: 121, 结束: 133标签: 结束时间, 文本: 'C)2502031600', 起始: 134, 结束: 146标签: 计划时间, 文本: 'D)JAN 29-31 FEB 01 03 0900-1600', 起始: 147, 结束: 177标签: 空域描述, 文本: 'E) AIRSPACE CLSD WI AREA:
554930N0384015E-554916N0383931E-
554903N0384027E-554902N0384104E-
554905N0384411E-554909N0384845E-
554940N0384840E-554930N0384512E-
554925N0384106E-554930N0384015E', 起始: 178, 结束: 336标签: 多边形坐标序列, 文本: '554930N0384015E-554916N0383931E-
554903N0384027E-554902N0384104E-
554905N0384411E-554909N0384845E-
554940N0384840E-554930N0384512E-
554925N0384106E-554930N0384015E', 起始: 204, 结束: 336标签: 限制的下边界, 文本: 'F)SFC', 起始: 336, 结束: 341标签: 限制的上边界, 文本: 'G)640M AMSL', 起始: 342, 结束: 353
候选 0 完成原因: FinishReason.STOP (STOP)