首页 / 文章

google-cloudragvertex-aigeminisupport-automation

Google Cloud上的双引擎RAG支持助手：公共文档、私有操作手册

开发一个支持助手，它能同时查询Vertex AI Search和开发者知识API，进而让Gemini将建议与客户的自定义规则进行协调。

发布者：REACTAPP.TOP阅读约 13 分钟2964 词

来源: How I built a dual-engine AI Support Assistant with Gemini Agent Search and Google Developer Knowledge API

任何在服务级别协议约束下处理支持请求的人都知道切换标签页的代价：一个用于查看 Google Cloud 的公开文档，另一个用于查阅客户自己的操作手册，还要在脑海中将两者信息整合起来。让通用人工智能助手帮忙简化这一流程时，得到的答案往往在理论上正确，但对眼前的客户来说却并不适用。

此处介绍的解决方案是一种双引擎检索增强型架构。工单会同时发送到两个来源，即公开的 Google Cloud 文档和客户私有的操作手册库，随后 Gemini 模型会将返回的信息整合成一份符合客户规则的解决方案指南。这就好比在网上找到一道评价很高的食谱，再请营养师根据你的饮食需求对其进行调整后再烹饪。

该系统由四个部分组成：

  • 两种并行运行的检索路径，一种用于提供通用技术建议，另一种则遵循严格的内部政策。
  • Vertex AI Search作为私有存储库，直接从操作手册文件中索引业务规则与约束条件。
  • Developer Knowledge API作为公共信息源，从官方文档中提供有依据的建议及故障排除步骤。
  • Gemini作为推理层，将公共解决方案与私有规则进行比对，从而生成支持工程师可安全遵循的指导方案。

为何通用建议会出错

以这样的工单为例：

Acme公司报告其Cloud Run前端存在大约五秒的冷启动延迟，我们该怎么做？

仅依据公开文档的助手几乎肯定会建议将min-instances设置为1。这是解决冷启动问题的标准方法，但对该客户而言也存在问题，原因有二：

  • 预算限制。Acme规定前端服务最多只能使用一个实例，因此无法使用始终在线的实例。
  • 流程规则。Acme的操作手册禁止在未经首席架构师莎拉·康纳书面批准的情况下更改min-instances值。

所需的是这样一种系统：它能获取有关Cloud Run启动延迟的公开指导，将每项建议与Acme的内部限制进行比对，只有在发现冲突时才告知工程师，并提出符合要求的替代方案。

前提条件

在开始之前请准备好以下内容：

  1. 已开启计费功能的 Google Cloud 项目。
  2. 已在您的机器上安装并通过身份验证的 gcloud CLI 工具。
  3. Python 3.10 或更高版本。

成本与配额

本教程旨在在各项服务的免费层级及开发者配额范围内进行操作：

  • Cloud Storage。 一个小的运行手册文件完全在永久免费存储额度之内，只有超出该额度或产生数据导出时才会产生费用。
  • Vertex AI Search（Gemini Agent Search）。 新账户可享受免费试用层级，试用额度用完后，每次查询将按 1,000 次为基准收取费用，具体价格请查看当前定价页面。
  • Developer Knowledge API。 启用和使用该服务均免费，且需在标准开发者限制范围内操作。配额使用情况可在 Google Cloud 的配额页面中查看。
  • Gemini 3.5 Flash。在Google AI Studio中，符合常规开发者使用限制即可免费使用。只有当您升级以获得更高的并发处理能力和速率限制时，才会开始按量计费。

    • 准备Google Cloud环境

    在启用助手功能之前，首先需要搭建底层基础设施。一个bash脚本可完成三项任务：

    • 启用API。它会开启Cloud Storage、Vertex AI Agent Builder（discoveryengine）、开发者知识库以及生成式语言API，这样调度器就能访问所需的各个端点。
    • 创建 Discovery Engine 服务身份。它会预先创建由 Google 管理的服务代理（service-PROJECT_NUMBER@gcp-sa-discoveryengine.iam.gserviceaccount.com），这样背景索引器即可立即获得 IAM 权限，无需首次在控制台进行操作。
    • 设置存储与访问权限。它会创建用于存放客户操作手册的 Cloud Storage 存储桶，并为两个主体授予 roles/storage.objectViewer 权限：一个是需要进行索引的背景搜索代理，另一个是您的账户，以便您能在控制台中查看该存储桶。

    将脚本粘贴到终端中：

    # 1. Initialize variables from active gcloud config
export PROJECT_ID=$(gcloud config get-value project)
export PROJECT_NUMBER=$(gcloud projects describe ${PROJECT_ID} --format="value(projectNumber)")
export BUCKET_NAME="${PROJECT_ID}-internal-kb"
export USER_EMAIL=$(gcloud config get-value account)

# 2. Enable all required GCP APIs
gcloud services enable \
  storage-api.googleapis.com \
  discoveryengine.googleapis.com \
  developerknowledge.googleapis.com \
  generativelanguage.googleapis.com \
  --project=${PROJECT_ID}

# 3. Provision the Discovery Engine Service Identity programmatically
# (This creates the background service account without requiring a manual Console click)
gcloud beta services identity create \
  --service=discoveryengine.googleapis.com \
  --project=${PROJECT_ID}

# 4. Create the GCS bucket
gcloud storage buckets create gs://${BUCKET_NAME} --project=${PROJECT_ID} --location=us-central1

# 5. Grant Storage Object Viewer to the Agent Search Service Agent (for background indexing)
gcloud storage buckets add-iam-policy-binding gs://${BUCKET_NAME} \
  --member="serviceAccount:@gcp-sa-discoveryengine.iam.gserviceaccount.com">service-${PROJECT_NUMBER}@gcp-sa-discoveryengine.iam.gserviceaccount.com" \
  --role="roles/storage.objectViewer"

# 6. Grant Storage Object Viewer to your active Console user (for frontend validation)
gcloud storage buckets add-iam-policy-binding gs://${BUCKET_NAME} \
  --member="user:${USER_EMAIL}" \

  --role="roles/storage.objectViewer"

echo "Setup complete! All APIs enabled and permissions configured successfully."

    您可能需要调整的参数如下：

    • PROJECT_ID 是您的项目编号。如果该项目已在 gcloud 中设置，gcloud config get-value project 会自动返回该值。
    • PROJECT_NUMBER 是项目的唯一编号。
    • BUCKET_NAME 用于为运行手册指定存储桶名称，默认值为 ${PROJECT_ID}-internal-kb。
    • USER_EMAIL 是可在控制台获得查看权限的活跃 Google Cloud 账户的地址。

    构建私有知识引擎

    私有端是一个小型搜索引擎，用于存储示例客户运行手册。

    编写并上传运行手册

    先在本地创建运行手册文件，然后将其复制到存储桶中：

    # Create the client runbook file with .txt extension
cat << 'EOF' > client-acme-runbook.txt
# Client Acme Corporation - Cloud Run Deployment Runbook
* **Client ID:** ACME-404
* **SLA Tier:** Platinum (4-hour resolution target)
* **Infrastructure Specifications:**
  * Frontend Service Name: `acme-frontend-prod`
  * Regional Deployment: `us-central1`
  * Max Instances: `1` (Hard cap imposed due to strict budget constraints)
  * CPU Allocation: `CPU allocated only during request processing`
  * Backend Database: Cloud SQL PostgreSQL (`acme-db-prod`)
  *   Networking: Serverless VPC (Virtual Private Cloud) Access connector (`acme-vpc-connector`)
* **Deployment Constraints:**
  * Deployed strictly via GitHub Actions workflow (`.github/workflows/deploy.yml`). Manual console updates are forbidden.
  * Direct access to the production DB requires using the Bastion Host `acme-bastion-prod`.
* **Escalation Path:**
  * Primary Contact: Sarah Connor (Lead Architect, sarah@acme-corp.com)
  * Secondary Contact: PagerDuty escalation policy `acme-prod-critical`
* **Common Issues & Custom Rules:**
  * **High Cold Starts:** High cold-start latency due to CPU throttling during scale-up. Do not modify the `min-instances` count without approval from Sarah Connor, as it breaches billing thresholds.
  * **Database Connection Timeouts:** If timeouts occur, verify that the Serverless VPC Access connector is in a `READY` state before attempting database restarts.
EOF

# Upload the file to your bucket
gcloud storage cp client-acme-runbook.txt gs://${BUCKET_NAME}/

    上传命令会使用设置步骤中指定的存储桶名称 ${PROJECT_ID}-internal-kb，除非您已更改过它。该运行手册刻意包含了所有关键信息：Acme规定的单个实例上限、最小实例数量审批规则、数据库及VPC连接器的名称、仅通过GitHub Actions进行部署的政策，以及应急联系信息。

    创建Agent搜索数据存储

    数据存储是让运行手册可检索的基础。在Google Cloud控制台操作如下：

    1. 打开Agent Builder（也标为Vertex AI Search）。
    2. 在左侧导航中选择Data Stores，然后点击Create data store。
    3. 作为数据源，选择Cloud Storage。
  • 对于文件夹路径，输入 gs://PROJECT_ID-internal-kb/*，并将 PROJECT_ID 替换为您实际的项目编号。
  • 选择非结构化文档（PDF、HTML、TXT等）作为数据类型，然后继续操作。
  • 为数据存储起名 acme-internal-kb。
  • 点击创建。
  • 在数据存储列表中，复制并记下数据存储ID。

    • 创建用于服务的搜索应用

    来自代码的查询会通过搜索应用（即引擎）处理，该引擎作为数据存储之上的服务层：

    1. 在Agent Builder中，打开应用选项。
    2. 点击创建应用。
    3. 在创建页面的搜索选项下，点击创建。
    4. 将此应用命名为 acme-search-app。
    5. 外部组织名称请输入 Acme Corporation。
    6. 位置保持默认值 global，然后继续操作。
    7. 在 Data Stores 步骤中，勾选 acme-internal-kb。
    8. 点击 Create。
    9. 操作完成后，从 Apps 列表中打开该应用，并在详情页复制其 Engine ID（也称为 App ID），格式类似 acme-search-app_1234567890。

    连接公共端：开发者知识 API

    该 API 可让用户便捷地访问官方的 Google Cloud 文档和开发者指南。以下步骤将用于创建 API 密钥、对其进行限制，并通过一次 curl 请求确认访问权限。

    创建 API 密钥

    1. 在控制台中，进入 Credentials。
    2. 选择创建凭证，然后选择API密钥。
    3. 建议将该密钥仅限制用于Developer Knowledge API。
    4. 将密钥存储在环境变量中：
    export DEVELOPERKNOWLEDGE_API_KEY="DEVELOPERKNOWLEDGE_API_KEY

    将您自己生成的密钥替换占位符。

    使用 curl 进行测试

    通过一次简单查询即可确认密钥有效，且您能够访问 Google 的开发者文档：

    curl -X POST "https://developerknowledge.googleapis.com/v1:answerQuery?key=$DEVELOPERKNOWLEDGE_API_KEY" \
-H "Content-Type: application/json" \
-d '{"query": "How do I reduce Cloud Run cold starts?"}'

    如果返回的是包含来自官方文档的答案的结构化 JSON，那就说明系统运行正常。

    协调脚本

    接下来是一个 Python 脚本，它会向两个引擎提出相同的问题，并将结果传递给 Gemini 以得出最终答案。

    安装库

    pip install google-cloud-discoveryengine google-generativeai requests

    创建协调器

    以下命令用于生成app.py。该脚本会将收到的工单文本发送到两个搜索路径，将两个搜索结果整合到提示模板中，然后将填充好的提示传递给Gemini：

    cat << 'EOF' > app.py
import os
import sys
import requests
import google.generativeai as genai
from google.cloud import discoveryengine_v1beta as discoveryengine
# Configuration - Replace with your details
PROJECT_ID = os.environ.get("PROJECT_ID")
ENGINE_ID = os.environ.get("ENGINE_ID")
LOCATION = "global"
MODEL_NAME = os.environ.get("GEMINI_MODEL", "gemini-3.5-flash")
# Initialize Gemini Client (Uses GEMINI_API_KEY env variable)
genai.configure(api_key=os.environ.get("GEMINI_API_KEY"))
model = genai.GenerativeModel(MODEL_NAME)

# 1. Helper to query the Private KB (Agent Search)
def query_private_kb(query_text: str):
client = discoveryengine.SearchServiceClient()
serving_config = f"projects/{PROJECT_ID}/locations/{LOCATION}/collections/default_collection/engines/{ENGINE_ID}/servingConfigs/default_search"
request = discoveryengine.SearchRequest(
serving_config=serving_config,
query=query_text,
page_size=3,
)
response = client.search(request)
results = []
for result in response.results:
doc_data = result.document.derived_struct_data
if "snippets" in doc_data:
for s in doc_data["snippets"]:
results.append(s.get("snippet", ""))
return "\n".join(results)

# 2. Helper to query the Public KB (Developer Knowledge API)
def query_public_gcp_kb(query_text: str):
api_key = os.environ.get("DEVELOPERKNOWLEDGE_API_KEY")
url = f"https://developerknowledge.googleapis.com/v1:answerQuery?key={api_key}"
headers = {"Content-Type": "application/json"}
payload = {"query": query_text}
try:
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
data = response.json()
return data.get("answer", {}).get("answerText", "No answer found.")
except Exception as e:
return f"Error querying Developer Knowledge API: {e}"

# 3. Main Orchestration
def run_support_assistant(ticket_query: str):
print(f"\n[Ticket Query]: {ticket_query}")
# Dynamically search private client specs using the ticket query
print("Searching private database…")
private_context = query_private_kb(ticket_query)
# Dynamically search public GCP advice using the ticket query
print("Searching public Google Cloud docs…")
public_context = query_public_gcp_kb(ticket_query)
# Format system prompt template
prompt = f"""
You are a Senior Cloud Support Assistant. Your task is to resolve the support ticket.
CRITICAL INSTRUCTIONS:
1. You MUST read the [Private Runbook Context] for specific client configurations, limits, and rules.
2. You MUST cross-reference all generic GCP recommendations against the client's private constraints. If a generic recommendation (e.g., increasing min-instances) violates a client limit or rule (e.g., budget caps or requiring specific manager approvals), you MUST call this out explicitly and provide the compliant fallback option.
3. Refer to specific client contacts (e.g., Sarah Connor), services, databases, and network connectors mentioned in the runbook when explaining the steps.
[Private Runbook Context]:
{private_context}
[Official GCP Guidance]:
{public_context}
[Support Ticket Query]:
{ticket_query}
Provide a precise troubleshooting guide for the support engineer.
"""
response = model.generate_content(prompt)
print("\n[AI Assistant Resolution Guide]:")
print(response.text)
if __name__ == "__main__":
if not all([PROJECT_ID, ENGINE_ID, os.environ.get("DEVELOPERKNOWLEDGE_API_KEY"), os.environ.get("GEMINI_API_KEY")]):
print("Please set your environment variables: PROJECT_ID, ENGINE_ID, DEVELOPERKNOWLEDGE_API_KEY, and GEMINI_API_KEY.")
else:
# Allow passing the ticket query as a command line argument
default_query = "Client Acme is complaining about 5-second cold start latency on their frontend service. How can we fix it?"
query = sys.argv[1] if len(sys.argv) > 1 else default_query
run_support_assistant(ticket_query=query)
EOF

    提示词才是真正起作用的部分。它指示模型读取针对特定客户设置的私有操作手册及规则，并将每一项通用建议与这些约束条件进行对比。如果某个公开的建议，比如增加最小实例数，违反了客户的限制或审批规则，模型必须明确指出这一点并提供符合要求的替代方案。在解释相关步骤时，模型还必须注明该客户的实际联系人、服务、数据库及网络连接信息。

    运行测试

    设置环境变量后启动脚本：

    export PROJECT_ID="PROJECT_ID"
export ENGINE_ID="ENGINE_ID"
export DEVELOPERKNOWLEDGE_API_KEY="DEVELOPERKNOWLEDGE_API_KEY"
export GEMINI_API_KEY="GEMINI_API_KEY"
# Run with the default query
python3 app.py
# Or pass a custom query as a command line argument
python3 app.py "Client Acme is complaining about database connection timeouts on their Cloud SQL backend."

    替换为实际值：

    • PROJECT_ID 是您的项目编号。
    • ENGINE_ID 是 Vertex AI Search 应用的引擎/应用编号。
    • DEVELOPERKNOWLEDGE_API_KEY 是您为开发者知识 API 生成的密钥。
    • GEMINI_API_KEY 是来自 Google AI Studio 的密钥。Google 的文档说明了如何获取 Gemini API 密钥。

    输出结果示例

    结果大致如下所示：

    [AI Assistant Resolution Guide]:
### **Acme Support Ticket: Resolution & Troubleshooting Guide**
**Ticket ID:** #ACM-98431
**Service:** `acme-frontend` (Cloud Run)
**Issue:** 5-second cold start latency
**Assigned To:** Cloud Support Engineering Team
 - -
### **CRITICAL COMPLIANCE WARNING (Read Before Acting)**
According to Acme's operational rules:
* **The Conflict:** Standard GCP documentation suggests setting `min_instances` (e.g., to 1 or more) to eliminate cold starts. However, **Acme operates under a strict budget cap policy that prohibits idle/always-on instances** for the frontend service without explicit business justification.
* **The Rule:** You **MUST NOT** increase `min_instances` on `acme-frontend` without prior written approval from **Sarah Connor** (Acme's Technical Lead & Budget Approver).
* **Compliant Fallback:** If Sarah Connor does not approve the budget increase for `min_instances`, you must resolve the latency strictly through **code-level optimizations, CPU adjustments, startup probes, and connection tuning** as detailed below.
 - -
### **Step-by-Step Troubleshooting & Resolution Guide**
#### **Phase 1: Triage and Governance Check**
1. **Contact Sarah Connor:** Reach out to Sarah Connor to request a budget exception for setting `min_instances = 1`.
2. **Fallback implementation:** If denied, proceed strictly with the budget-compliant fallback steps below to optimize the cold start duration within the 0-instance scaling limit.
#### **Phase 2: Network & Connection Optimization**
1. **Inspect VPC Connector Latency:** `acme-frontend` routes traffic to the database via the **`acme-vpc-connector`** (Serverless VPC Access). Ensure that the connector throughput is not bottlenecked.
2. **Database Connection Lazy-Loading:** The frontend connects to **`acme-production-db`** (Cloud SQL). Do not open the connection pool until the first actual handler request arrives to minimize container boot time.
…

    结果显示的内容

    答案结合了两个信息来源，每个来源提供不同的内容：

    1. 通过开发者知识 API 提供的公开文档，列出了技术解决方案：启用 cpu-boost、将并发数设置为 80，或添加 /healthz 启动检测。
    2. 通过 Agent Search 访问的私有知识库，补充了业务逻辑、安全约束及操作细节。它指出标准建议（min_instances = 1）违反了 Acme 的预算规则，并说明必须由 Sarah Connor 批准任何例外情况。此外还提到了 Acme 的实际数据库和 VPC 连接器。

    如果没有这两种引擎，普通模型只会建议提高min_instances的值。初级支持工程师可能会立即照做，而在企业环境中这可能会导致预算超限、触发计费警报，并跳过架构审批流程。将模型置于公共和私有环境之中，才能既得到技术上合理的解决方案，又能符合组织政策要求。

    为生产环境做强化处理

    在用于处理实际工单之前，做一些调整是必要的：

    • 将敏感信息存储在密钥管理器中。应将API密钥存放在密钥管理器中，并在运行时加载，而非以明文形式存在环境变量中。
  • 使用最小权限的 IAM 权限。在仅拥有 roles/discoveryengine.viewer 和 roles/storage.objectViewer 权限的专用服务账户下运行工作负载，并将 API 密钥限制在特定的 API 和可信网络地址上。
  • 实现自动数据导入。通过 Vertex AI Search 连接器将 Jira、Confluence、SharePoint 或 Google Drive 等数据源与数据存储库关联，从而无需手动上传。
  • 通过元数据过滤。在 SearchRequest 过滤条件中传入客户端 ID 和系统标签，以便将搜索结果限定在特定客户的环境中。
  • 缓存常用查询结果。将频繁出现的文档答案存储在 Memorystore for Redis 中，以降低延迟并减少配额消耗。

    • 清理工作

    为避免持续产生费用，请删除教程中创建的所有内容。

    搜索应用与数据存储

    当有应用关联时无法删除数据存储，因此需先删除该应用。在控制台操作如下：

    1. 进入Agent Builder，然后选择Apps。
    2. 找到acme-search-app，点击更多操作（三个点）并选择删除。
    3. 在确认对话框中输入应用名称并确认。
    4. 在导航菜单中打开Data Stores。
    5. 找到acme-internal-kb，点击更多操作，然后选择删除。
    6. 确认操作。

    存储桶

    清除桶内的内容只需一条命令：

    export BUCKET_NAME="${PROJECT_ID}-internal-kb"
gcloud storage rm - recursive gs://${BUCKET_NAME}/

    API密钥

    1. 进入APIs & Services，然后选择Credentials。
    2. 在API Keys下方找到为Developer Knowledge API生成的密钥。
    3. 点击其旁边的垃圾桶图标并确认删除。

    本地文件

    从您的机器上删除脚本及示例运行手册：

    rm -f app.py app_test.py client-acme-runbook.txt
    ← 全部文章发送简报