并行 Monitor 集成

使用 LangChain Python 与 ParallelMonitor 类型集成。

⟩Parallel 是为 LLMs 和 AI 应用构建的实时网页搜索与内容提取平台。

ParallelMonitor 调用 Parallel 的 Monitor API 按周期性频率（1 小时至 30 天）安排查询，并通过 Webhook 或轮询通知您新内容出现。适用于持续竞争情报追踪、监管动态监控或主题监测。

概览

集成细节

类别	包名	可序列化	JavaScript 支持	最新包版本
@[ParallelMonitor]	@[langchain-parallel]	❌	❌

安装

集成模块位于 langchain-parallel 包中。

pip

pip install -U langchain-parallel

凭据

前往 Parallel 注册并生成 API 密钥。将 PARALLEL_API_KEY 设置为环境变量：

import getpass
import os

if not os.environ.get("PARALLEL_API_KEY"):
    os.environ["PARALLEL_API_KEY"] = getpass.getpass("Parallel API key:\n")

实例化

from langchain_parallel import ParallelMonitor

client = ParallelMonitor()

创建监控

frequency 是 <n><unit>，单位为 h、d 或 w，最终持续时间存储在 [1h, 30d]（例如 "1h"、"6h"、"3d"、"1w"、"2w"）。

from langchain_parallel import MonitorWebhook

monitor = client.create(
    query="Latest peer-reviewed papers on net-energy-gain fusion",
    frequency="1d",
    metadata={"label": "fusion-watch"},
    # Optional: receive new events on a webhook endpoint.
    # webhook=MonitorWebhook(
    #     url="https://example.com/parallel/webhook",
    #     event_types=["monitor.event.detected"],
    # ),
)

print(monitor["monitor_id"], "—", monitor["status"])

有效的 event_types 是 monitor.event.detected、monitor.execution.completed 和 monitor.execution.failed。省略该字段将返回三者结果。

查询、列出、删除

got = client.retrieve(monitor["monitor_id"])

monitors = client.list(limit=20)

deleted = client.delete(monitor["monitor_id"])

列出事件

list_events() 返回一个扁平的 {"events": [...]} 列表。新创建的监控器在首次计划执行前通常没有事件；调用方式无论哪种情况都适用。

events = client.list_events(monitor["monitor_id"], lookback_period="7d")
for ev in events.get("events", []):
    print(ev["type"], ev.get("event_date"))

To inspect a single event group:

group = client.get_event_group(monitor["monitor_id"], event_group_id="...")

模拟事件

在配置webhook处理器时很有用。simulate_event() 会向监控器的webhook发送请求类型（默认为 monitor.event.detected）的合成事件。

simulate_event() 需要在监控器上配置webhook。首次调用 create() 时传递 webhook=MonitorWebhook(...)；否则API会返回 400 — Webhook not configured for this monitor。

client.simulate_event(monitor["monitor_id"], event_type="monitor.event.detected")

异步

每個CRUD和事件方法都有a前缀的异步對應方法：

monitor = await client.acreate(query="...", frequency="6h")
events = await client.alist_events(monitor["monitor_id"], lookback_period="7d")
await client.adelete(monitor["monitor_id"])

Webhook簽名驗證

Webhook递交物使用標準Webhook方案（HMAC-SHA256過<webhook-id>.<webhook-timestamp>.<body>，base64編碼的v1,<sig>格式的驗證，並包含重放保護）。驗證方式如下verify_webhook：

from langchain_parallel import verify_webhook

ok = verify_webhook(
    body,
    webhook_id=request.headers["webhook-id"],
    webhook_timestamp=request.headers["webhook-timestamp"],
    webhook_signature=request.headers["webhook-signature"],
    secret=os.environ["PARALLEL_WEBHOOK_SECRET"],
)

簽名密鑰在Parallel控制臺的組Webhook端點级别配置，而非按監控器级别配置。請參考webhook setup以查看完整的交付合同。

响應格式

create()和retrieve()返回monitor字典。list_events()返回扁平的事件封裝。

# create() / retrieve()
{
    "monitor_id": "mon_abc",
    "status": "active",
    "query": "...",
    "frequency": "1d",
    "metadata": {...},
    "webhook": {"url": "...", "event_types": [...]},
    "created_at": "2026-04-15T10:30:00Z",
    # ... additional API-side fields
}

# list_events()
{
    "events": [
        {
            "type": "event",            # or "completion" / "error"
            "event_date": "2026-04-15T10:30:00Z",
            "event_group_id": "...",
            # ... event-type-specific fields
        },
    ],
}

API參考

詳細文檔請前往@[ParallelMonitor] API參考或Monitor API guides。

---

官方原文：src/oss/python/integrations/tools/parallel_monitor.mdx · 非官方翻译，译文由 AI 生成，以英文原文为准。