｜Agentic AI Group
智能体工程课程
2026 版
进阶10 学时更新：2026-07-26

07｜外部 API、鉴权、数据库与知识图谱

将智能体安全接入第三方 API、结构化数据库、向量库、缓存和知识图谱，建立可恢复、可审计的跨系统调用链。

前置知识
  • 了解 HTTP、JSON、SQL 和基本身份认证
  • 完成自定义 Tool 章节
学习成果
  • 能实现带超时、重试和参数映射的异步 API Tool
  • 能正确区分 API Key、Bearer Token、OAuth 与签名鉴权
  • 能用受控查询接入 SQL、缓存和知识图谱
智能体跨系统集成链路
图 7-1：跨系统动作通过网关和适配层统一鉴权、可靠性与审计。

1. API 集成与异步调用

1.1 先定义系统契约

接入外部 API 前明确：请求地址、方法、参数、返回 Schema、认证方式、限流、超时、重试、幂等、数据分类、错误码和服务等级。不要让模型直接拼接任意 URL 或 HTTP 请求。

1.2 参数映射

模型生成的是业务参数，适配器转换为外部接口格式：

from dataclasses import dataclass

@dataclass
class WeatherRequest:
    city: str
    date: str


def to_provider_payload(request: WeatherRequest) -> dict:
    return {
        "location": request.city.strip(),
        "forecast_date": request.date,
        "units": "metric",
    }

映射层负责默认值、枚举转换、字段脱敏和兼容不同版本，模型不需要知道供应商细节。

1.3 异步、超时与重试

import asyncio
import httpx

TRANSIENT = {408, 429, 500, 502, 503, 504}

async def call_api(url: str, payload: dict, token: str) -> dict:
    timeout = httpx.Timeout(connect=2.0, read=8.0, write=5.0, pool=2.0)
    headers = {"Authorization": f"Bearer {token}"}
    async with httpx.AsyncClient(timeout=timeout) as client:
        for attempt in range(3):
            response = await client.post(url, json=payload, headers=headers)
            if response.status_code < 400:
                return response.json()
            if response.status_code not in TRANSIENT:
                response.raise_for_status()
            await asyncio.sleep(0.5 * (2 ** attempt))
    raise TimeoutError("外部服务连续失败")

重试只适用于临时错误。对写操作，必须配合幂等键和结果查询，避免重复执行。

2. API Key、OAuth、签名与密钥治理

2.1 四类常见方式

方式适用主要风险关键措施
API Key服务到服务、简单调用泄露后可直接使用最小权限、限 IP、轮换
Bearer TokenOAuth 资源访问持有者即可使用TLS、短有效期、安全存储
OAuth 2.0代表用户访问资源回调劫持、Token 滥用PKCE、精确回调、状态校验
请求签名高安全服务间调用时钟、重放、密钥泄露时间戳、Nonce、验签与轮换

OAuth 2.0 的目标是让第三方获得有限访问，而不是获得用户密码。当前安全实践还应参考 OAuth 2.0 Security Best Current Practice（RFC 9700）。

2.2 密钥治理

  • 不写入 Prompt、代码仓库、日志或前端；
  • 使用环境变量仅适合本地开发，生产使用密钥管理服务；
  • 为不同环境和服务分配不同凭据；
  • 定期轮换，旧密钥有短暂重叠期；
  • 记录密钥 ID，不记录密钥值；
  • 凭据泄露后可快速吊销和追踪影响范围。
禁止做法让模型读取全部环境变量，再自行挑选密钥。模型上下文不是密钥保险箱。

3. 结构化数据库、向量库与缓存

3.1 SQL Tool 的安全边界

最安全方式是暴露参数化业务函数，而不是任意 Text-to-SQL：

import sqlite3


def get_order(db: sqlite3.Connection, order_id: str, user_id: str) -> dict | None:
    row = db.execute(
        """
        SELECT order_id, status, amount
        FROM orders
        WHERE order_id = ? AND owner_id = ?
        """,
        (order_id, user_id),
    ).fetchone()
    if row is None:
        return None
    return {"order_id": row[0], "status": row[1], "amount": row[2]}

必须使用参数化查询，限制表、列、行数、执行时间和只读权限。需要 Text-to-SQL 时，增加 SQL 解析、白名单、查询计划检查和人工审批。

3.2 向量数据库

向量数据库用于语义检索，但仍需元数据过滤、多租户隔离、版本和删除。Embedding 输入不应包含无必要的敏感字段。

3.3 缓存

可缓存：模型分类结果、稳定检索结果、Embedding 和只读工具结果。缓存 Key 应包含模型/Prompt/索引版本和权限上下文，防止旧结果或跨用户泄露。

4. 知识图谱接入

4.1 构建流程

  1. 定义实体和关系 Schema；
  2. 从结构化数据或文本抽取候选；
  3. 实体归一化与消歧；
  4. 加入来源、时间和置信度；
  5. 人工或规则校验高价值关系；
  6. 写入图数据库并建立查询模板。

4.2 图谱查询 Tool

ALLOWED_RELATIONS = {"BELONGS_TO", "DEPENDS_ON", "USES"}


def build_relation_query(entity_id: str, relation: str) -> tuple[str, dict]:
    if relation not in ALLOWED_RELATIONS:
        raise ValueError("不允许的关系类型")
    cypher = f"""
    MATCH (a {{id: $entity_id}})-[r:{relation}]->(b)
    RETURN b.id AS id, b.name AS name
    LIMIT 20
    """
    return cypher, {"entity_id": entity_id}

关系类型来自白名单，实体值使用参数，避免模型生成任意图查询。

4.3 图谱与 RAG 融合

用户问“某设备故障会影响哪些生产线”时，可先通过图谱找到设备—产线关系，再检索相关维护文档，最后把结构关系和文本证据一并输出。

5. 跨系统安全与异常治理

5.1 可靠性模式

  • Timeout：所有远程调用必须有超时；
  • Retry：只重试临时错误，指数退避并加随机抖动；
  • Circuit Breaker：连续失败时快速拒绝，防止雪崩；
  • Bulkhead：不同工具使用独立资源池；
  • Fallback：切换只读、缓存或人工流程；
  • Idempotency：写操作可安全重放；
  • Compensation：长事务失败时执行补偿动作。

5.2 日志与追踪

日志中保存 trace_id、工具名、调用方、脱敏参数摘要、状态码、耗时和重试次数。禁止记录访问 Token、身份证号、完整病历等敏感信息。

6. 应用案例：跨系统工单助手

系统接收故障描述，查询设备数据库、检索 SOP、查看历史工单、生成处理建议，经人工确认后创建新工单。

6.1 工具列表

  • get_equipment(equipment_id)：只读 SQL；
  • search_sop(query, acl)：权限过滤 RAG；
  • find_similar_tickets(summary)：向量检索；
  • create_ticket(payload, idempotency_key)：写 API，需要审批；
  • notify_owner(ticket_id)：消息 API。

6.2 故障演练

  1. 工单 API 返回 429；
  2. 设备数据库超时；
  3. 用户无权查看某生产线；
  4. 模型尝试创建重复工单；
  5. 日志中出现敏感字段。

学生需要为每个故障给出系统行为、用户提示、审计事件和恢复方式。

6.3 工程复现：跨系统工单事务

cd repository/enterprise-agent-lab
python -m app.cli --question "请创建维修工单" --user guest --approve
python -m app.cli --question "请创建维修工单" --user employee-001 --approve
python -m unittest discover -s tests -v
python -m app.server --host 127.0.0.1 --port 8080

app/tools.py 与 data/equipment.json/tickets.json 是本章适配层。先验证 guest 无写权限，再验证正常写入和幂等。将模拟 API 替换为真实系统前，必须定义 OAuth audience/scope、连接/读取超时、仅对 429/临时 5xx 的重试、熔断、补偿和脱敏日志。

部署按 DEPLOY.md，密钥只由服务器 Secret 管理注入。验收要执行数据库超时、API 429、过期 Token、权限不足、重复请求五个故障，并为每个故障展示用户提示、审计事件、恢复方式和最终业务状态。

7. 可靠参考资料