接入外部系统：DB / REST API / 文件系统（以 USGS 地震快报 Demo 为例）

本文是智能体系列文章的一篇，聚焦“外部系统接入”这一工程化分水岭：如何用 REST 获取事实、用 DB 管理状态、用 文件系统交付与审计，把一个可复现的闭环跑通。
示例工程：USGS 公共 GeoJSON + SQLite + 本地文件输入/输出（可选 DeepSeek 仅用于可读性增强）。

1. 引言：为什么“接入外部系统”是智能体落地的分水岭

很多团队第一次做智能体，容易把重点放在“模型能不能聊得更像人”。但在真实工程里，能聊往往只是门槛，真正决定智能体是否“落地可用”的，是它能不能稳定、可控地接入外部系统：拉到可信的数据、执行可验证的动作、并把结果以可交付的形式沉淀下来。

换句话说，智能体的本质不是“更强的聊天机器人”，而是一个具备感知—决策—执行—交付闭环的工程系统。闭环中最容易被忽略、却最决定成败的部分，就是“执行”和“交付”，而这两件事几乎必然依赖外部系统：业务数据在数据库里、实时状态在服务接口里、输入配置与交付物在文件系统或对象存储里。脱离外部系统的智能体，即便回答再流畅，也很难从“演示”走到“生产”。

因此，本文把“接入外部系统”视为智能体工程化的分水岭：

• 没有外部接入：智能体只能“说”，无法“做”；无法闭环，也无法验真。
• 具备外部接入能力：智能体可以“拿到事实、改变状态、产出交付物”，并且这些行为能够被审计、回放与复现，才有机会成为业务链路的一部分。

在工程实践中，可把外部系统接入简化为一个非常实用的“三件套”：

1) REST：拿事实、触达服务

REST API 是智能体获取实时信息、调用业务能力的最常见方式。比如拉取监控指标、读取工单详情、提交变更单草稿、调用第三方数据源等。REST 接入不只是“发 HTTP”，而更在超时与重试、退避策略、分页与限流、响应字段漂移、错误码处理、以及如何把不稳定的网络世界变成稳定的上层语义这些工程细节。

2) DB：管状态、做幂等、留证据

数据库是智能体从“脚本”走向“系统”的关键。它承担三类职责：

• 状态：一次任务的中间结果、历史记录、运行元信息；
• 幂等：重复运行不重复插入、外部数据更新可覆盖；
• 追溯：能够解释“为何做出该判断/该输出”，并能在事后复现。

不做 DB，很多看似简单的功能（比如“每天生成一次日报”“只处理新增事件”“失败重试不重复写入”）都会变得脆弱且不可控。

3) 文件系统：输入配置、交付输出、审计回放

文件系统（或对象存储）是智能体工程里最便宜、最直观、最可复现的交付通道：

• 输入侧：把规则、关注对象、参数做成配置文件，避免硬编码；
• 输出侧：把结果持久化为 Markdown/JSON/CSV 等，方便人读、也方便系统二次消费；
• 审计侧：保留原始响应、运行指标、关键中间件输出，确保出现争议或故障时能回放定位。

这也是为什么许多“看起来很智能”的 demo 一旦上线就不可用：它缺少可追溯的输入/输出与证据链。

综上，所谓“外部系统接入”，并不是让智能体多连几个接口这么简单，而是为智能体建立一套可执行、可验证、可追溯的工程骨架：

• 可执行：能调用外部能力完成任务；
• 可验证：关键决策能对齐事实数据与明确规则；
• 可追溯：输入、原始数据、处理过程与输出都有据可查。

围绕这个目标，本文使用一个尽量“少依赖、易复现”的示例工程，把三件套跑成闭环：调用公开 REST 拉取地震数据、使用 SQLite 维持状态与幂等、利用文件系统输出告警与日报并保留审计原文。

2. Demo 与工程目标

本文的示例工程要演示的不是“地震科普”，而是一个最小却完整的外部系统接入闭环。因此 Demo 的选型标准只有三条：公开可复现、结构足够典型、能自然覆盖工程关键点。

2.1 为什么选 USGS 公共 GeoJSON

本例选择 USGS（美国地质调查局）提供的地震信息公开接口，原因直接且直接：

• 公开稳定、无需 Key：读者不需要注册账号、申请密钥或搭建服务，就能跑通 REST 接入链路，最大化降低复现门槛。
• 数据结构典型：USGS 返回的是 GeoJSON（半结构化 JSON）。必须做解析、容错与规范化，才能把数据稳定地接进系统。
• 工程点齐全：拉取、持久化审计、入库去重、规则告警、报告输出，一个闭环里需要的关键组件都能自然出现。

示例默认使用 USGS 的 24 小时 feed（无需参数、最稳）来保证示例的确定性：过去 24 小时全部事件（.../summary/all_day.geojson）。当然也可扩展为 query API，但那会引入更多参数解释与边界情况处理，不适合本文的“最小示例”目标。

2.2 Demo 的工程目标：一键跑通“三件套”闭环

为了让读者把注意力放在“接入套路”而不是“框架选择”，本工程明确追求以下目标：

1. 一键运行：执行一次命令即可拉取数据、写库、生成产物。
2. 幂等可重复：同一套代码重复运行不会把数据写乱，不会无限增长重复记录。
3. 可追溯可回放：出现问题能定位原因；输出能解释“数据从哪里来、如何处理、如何生成”。
4. 依赖极简：不引入外部中间件，DB 采用 SQLite，本地即可运行。

这套目标最终落实为一条清晰流水线：

拉取（REST）→ 标准化（Model）→ 入库幂等（DB）→ 规则告警 → 报告持久化（FS）

其中，“标准化（Model）”是关键转折点：外部 API 的半结构化数据必须被规范为稳定对象，后续 DB、规则、输出才有一致的输入。

2.3 最小结构与关键产物

• 输入：data/regions.json（关注区域与阈值）
• 状态：data/quakes.db（SQLite）
• 交付：out/alerts.json（机器可读）、out/daily_digest.md（人可读）
• 审计/可观测：out/raw_usgs_*.json、out/metrics.json

有了这个定位，后续每一节都会围绕一个问题展开：如何把对应外部系统接入得更稳定、可控、可追溯。下文先从最容易被轻视、但最先出故障的 REST 开始。

3. REST 接入：拉取公开 API 的工程要点

在“接入外部系统”这件事上，REST 往往是最先遇到、也最容易被低估的一环。很多示例项目把它写成一行 requests.get(url).json()，看起来简单，但一旦进入真实运行环境，通常会立刻面临三个问题：不稳定（网络与对端波动）、不可控（超时与重试策略缺失）、不可追溯（当时到底返回了什么说不清）。

本示例的 REST 接入刻意只做“关键工程点”，并把这些点收拢到一个独立模块中（src/usgs_client.py），其目的很明确，即把 HTTP 的不确定性隔离起来，让主流程只处理“拿到数据之后”的确定性逻辑。

3.1 为什么要封装客户端：隔离不确定性边界

把 REST 拉取封装成 UsgsClient.fetch_all_day()，不是为了“面向对象好看”，而是为了建立清晰边界：

• 上层（main.py）只关心一件事：能否拿到一个合法的 GeoJSON（dict）
• 下层（UsgsClient）负责处理所有不确定性：超时、重试、退避、HTTP 错误、JSON 解析失败、结构校验

下面是本示例客户端方法的核心结构（已去除无关细节，仅保留关键逻辑）：

# src/usgs_client.py（节选）
def fetch_all_day(self, url: str = USGS_ALL_DAY_URL, max_retries: int = 2) -> Dict[str, Any]:
    attempts = 1 + max_retries  # 总尝试次数：首次 + 重试次数

    for i in range(1, attempts + 1):
        try:
            with httpx.Client(timeout=self._timeout, follow_redirects=True) as client:
                resp = client.get(url, headers={"Accept": "application/geo+json, application/json"})
                resp.raise_for_status()          # 4xx/5xx 显性化
                data = resp.json()               # JSON 解析（可能抛异常）

                if not isinstance(data, dict) or "features" not in data:
                    raise ValueError("USGS 响应缺少 features 字段或类型异常")

                return data                       # 上层只接收“合法的 dict”
        except (httpx.HTTPError, ValueError) as e:
            if i < attempts:
                sleep_backoff(i)                 # 指数退避
                continue
            raise

3.2 超时：默认必须有，且要“短而明确”

没有超时的 HTTP 调用，在自动化任务里几乎等同于“埋雷”。示例中把超时作为初始化参数显式传入（默认 10s），其体现的思想是任务必须具备可控的失败路径。

# main.py（节选）
client = UsgsClient(timeout_s=10.0)
data = client.fetch_all_day(url=args.url, max_retries=2)

3.3 重试与退避：要有，但不能“盲目连打”

重试解决的是短暂性失败，但没有退避会扩大故障影响面。本示例采用“有限次数 + 指数退避”，并把退避逻辑独立出来置于 src/utils.py之中：

# src/utils.py（节选）
def sleep_backoff(attempt: int, base: float = 0.5, cap: float = 6.0) -> None:
    """指数退避睡眠：min(cap, base * 2^(attempt-1))，attempt 从 1 开始。"""
    secs = min(cap, base * (2 ** (attempt - 1)))
    time.sleep(secs)

3.4 HTTP 状态与错误处理：先让错误“显性化”

示例通过 raise_for_status() 保证错误显性化，再由重试机制处理短暂错误。这其中的关键原则是失败要失败得清楚，不要吞掉错误继续生成“空报告”。

resp = client.get(url, headers={"Accept": "application/geo+json, application/json"})
resp.raise_for_status()
data = resp.json()

3.5 响应校验：不要把“错误数据”带进后续链路

即便收到的响应是 HTTP 200，也不代表数据就可用。示例做了最小校验：必须是 dict 且包含 features。

data = resp.json()
if not isinstance(data, dict) or "features" not in data:
    raise ValueError("USGS 响应缺少 features 字段或类型异常")

3.6 原始响应持久化：REST 不确定性的“证据链”

拿到响应后立即持久化 out/raw_usgs_<run_id>.json，支持审计、回放与复现，默认开启、必要时再关闭。

# main.py（节选）
run_id = str(uuid.uuid4())
raw_path = os.path.join(args.out, f"raw_usgs_{run_id}.json")

data = client.fetch_all_day(url=args.url, max_retries=2)

if not args.no_raw:
    write_json(raw_path, data, indent=2)

本节 REST 接入只做四件事：客户端封装、超时+重试+退避、状态码与结构校验、原始响应持久化。它们共同保证了 上层拿到的是“可信输入”，而不是不可控的网络噪声。下一步就是把可信输入固化成“可重跑的系统状态”。

4. DB 接入：SQLite 也能把“工程化”讲清楚

很多人一听到“数据库接入”，第一反应是会不会把示例做得太重？但在智能体工程里，DB 的价值并不取决于数据库有多“重”，而取决于它能否把三件事（即状态管理、幂等执行、可追溯证据链）落地。SQLite 恰好能用零外部依赖把这些要点讲清楚。

4.1 为什么必须有 DB：从“脚本”走向“系统”

通过 REST 拉到的数据是瞬时的。如果没有 DB，将很难回答以下问题：重复运行会不会重复写入？外部事件更新了如何合并？出现争议时能否复盘当时状态？而 DB 在本例中承担的角色实现了把外部不确定输入转化为内部可控、可复盘的状态。

4.2 自动建表：让工程可复现、可迁移

示例把建表逻辑放在 init_db() 中，运行时自动创建表，读者无需预置环境。

# src/store.py（节选）
def init_db(db_path: str) -> None:
    conn = sqlite3.connect(db_path)
    try:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS quake_event (
              event_id TEXT PRIMARY KEY,
              time_ms INTEGER NOT NULL,
              updated_ms INTEGER NOT NULL,
              mag REAL,
              place TEXT,
              url TEXT,
              tsunami INTEGER,
              felt INTEGER,
              usgs_alert TEXT,
              lat REAL NOT NULL,
              lon REAL NOT NULL,
              depth_km REAL,
              raw_json TEXT,
              inserted_at TEXT NOT NULL
            )
        """)
        conn.commit()
    finally:
        conn.close()

说明：这里仅展示 quake_event 表的建表过程；源码中同一函数还会初始化 quake_tag 与 run_log两张表。

4.3 Upsert 幂等：外部数据接入的“必修课”

自动化任务几乎一定会被重复执行，如果没有幂等策略，结果要么数据重复、要么状态错乱。本示例用 ON CONFLICT(event_id) DO UPDATE 实现 upsert，达到不存在则插入、存在则更新的效果。

# src/store.py（节选）
sql = """
INSERT INTO quake_event (...)
VALUES (...)
ON CONFLICT(event_id) DO UPDATE SET
  updated_ms=excluded.updated_ms,
  mag=excluded.mag,
  place=excluded.place,
  ...
"""
inserted_at = now_iso()

上述代码中有一个刻意的细节是使用inserted_at 保留记录首次插入时间，不随更新刷新，这样的设计有助于解释“事件什么时候进入系统”。

4.4 最小运行追溯：让一次运行“说得清楚”

示例提供了轻量的 run_log，把一次运行与 raw 输入、统计信息关联起来。它不是为了做复杂监控，而是为了让系统具备最基本的说明能力。

# src/store.py（节选）
def log_run_start(db_path: str, run_id: str, started_at: str, raw_path: str) -> None:
    conn = sqlite3.connect(db_path)
    try:
        conn.execute("""
            INSERT INTO run_log(run_id, started_at, raw_path)
            VALUES (?, ?, ?)
            ON CONFLICT(run_id) DO NOTHING
        """, (run_id, started_at, raw_path))
        conn.commit()
    finally:
        conn.close()

说明：默认情况下运行会写入 run_log；若运行时使用 --dry-run参数，则不会写入运行日志。

当 REST 的输入被固化为 DB 状态后，后续的规则判断与输出生成就能保持稳定与可复现。接下来进入到文件系统，实现真正交付结果，并补齐证据链。

5. 文件系统接入：输入配置 + 输出交付物 + 可追溯审计

文件系统在本章中承担的角色更接近“交付与证据链”，它实现智能体的行为可配置、产物可消费、过程可追溯。本示例把文件系统的作用拆成三类：输入配置、输出交付物、可追溯审计。

5.1 输入配置：规则可改、可版本化

data/regions.json 用于描述应用程序所关注的区域与阈值。其意义在于规则不硬编码在代码里，便于修改、对比、回滚，并可对输出进行解释与审计。

# main.py（节选）
cfg = read_json(args.regions)  # data/regions.json

5.2 输出交付物：同时面向机器与人

本示例输出的交付物主要有两个：

• out/alerts.json：机器可读，方便告警平台/消息推送/工作流二次集成
• out/daily_digest.md：人可读，便于传播与存档（知识库/Wiki/Git）

其实现代码如下：

# main.py（节选）
alerts = build_alerts(alert_candidates, tags_map, dist_map, max_alerts=max_alerts)
write_json(os.path.join(args.out, "alerts.json"), alerts, indent=2)

md = render_markdown(stats, top_events, region_hits, tags_map, dist_map)
write_text(os.path.join(args.out, "daily_digest.md"), md)

5.3 可追溯审计：raw 持久化 + 运行指标

外部系统接入排障，除了查看代码实现逻辑，更重要的是看“当时发生了什么”。因此示例默认输出两类证据文件：

• out/raw_usgs_*.json：原始响应审计，支持回放与复现
• out/metrics.json：最小可观测性，记录耗时、数量、产物路径等

说明：示例运行时默认使用持久化，可通过运行参数 --no-raw 关闭它。

# main.py（节选）
write_json(raw_path, data, indent=2)

metrics = {
    "run_id": run_id,
    "elapsed_s": round(time.time() - t0, 3),
    "fetched_features": len(features),
    "parsed_events": len(events),
    "parse_errors": parse_errors,
    "inserted_count": inserted_count,
    "alert_count": len(alerts),
}
write_json(os.path.join(args.out, "metrics.json"), metrics, indent=2)

到这里，三类接入点形成闭环：REST 拉取事实并固化 raw 审计，DB 以 upsert 管理状态与去重，文件系统输出可消费产物并持久化运行指标。

6. 决策边界

本示例刻意把“决策”划出清晰边界：是否告警、如何打标签等关键判断必须由可解释、可测试、可复现的规则来决定（例如震级阈值 min_magnitude_alert、USGS 自带告警颜色、与关注区域的距离命中等），并将这些规则固化为配置与代码逻辑，以保证重复运行结论一致、出现争议可追溯；相对地，DeepSeek（若启用）只用于提升可读性——比如为日报生成一段中文概述或对结果做语言组织——且必须支持“失败回退”（调用失败不影响主流程与产物生成）。换句话说：规则负责正确性与确定性，LLM 负责表达与体验，两者职责不混淆，这样才能把智能体从演示推向可上线的工程系统。

7. 小结：抽象成可复用模板

本文用一个“USGS 地震快报”的最小工程，把智能体落地时最常见、也最关键的外部系统接入闭环跑了一遍。更重要的不是这个场景本身，而是它背后可迁移的模板：Fetch → Persist → Deliver。

7.1 三段式模板：Fetch / Persist / Deliver

1. Fetch（REST 获取事实）
   目标不是“发出 HTTP 请求”，而是把不确定性控制在可接受范围内：明确超时、有限重试与退避、状态码显性化、结构校验，并保留原始响应作为证据。这样上层逻辑处理的是“可信输入”，而不是“可能是错误页的 200”。

2. Persist（DB 固化状态并保证幂等）
   REST 拿到的是瞬时数据，系统要可运行、可重跑、可追溯，就必须把关键事实落到状态存储中，用主键锚定实体、用 upsert 保证幂等、用最小运行记录把一次执行与输入/输出关联起来。状态被固化后，规则判断与输出才稳定。

3. Deliver（文件系统交付产物与审计）
   交付不是 print()，而是面向机器与人的双通道输出，即结构化告警（便于集成）+ 可读日报（便于传播）；同时用 raw 与 metrics 建立最小证据链与可观测性，让系统出了问题“说得清楚、查得到”。

7.2 迁移指南：换 API / 换表 / 换输出，套路不变

如果将本示例迁移到其他业务场景，通常只需替换三类组件，而不需要推倒重来：

• 换 REST：把 USGS 换成企业服务。保留超时/重试/校验/审计持久化这套“抗不确定性”机制，并可能需要增加鉴权等机制。
• 换 DB 结构：把 quake_event 等换成业务实体表。保留主键锚定 + upsert 幂等 + 最小运行追溯等策略，并可能需要更换 DB 选型以适应大数据量需求。
• 换交付物：把日报/告警换成 CSV/HTML/PDF/Webhook。保留“面向机器 + 面向人 + 可追溯证据”的交付理念。

7.3 一句话总结

智能体想落地，必须先把外部系统接入做成工程——让输入可控、状态可复盘、输出可交付。
当具备这套可复用模板后，后续再引入更复杂的能力（鉴权、限流、调度、编排、多智能体协作）都可以在这个骨架上增强，而不必每次都重新开始。

本文示例代码仓库：

https://gitcode.com/gtyan/AgentHandBook/tree/main/09

关注我，阅读Agent系列文章