数据源集成

AI智能体的能力很大程度上取决于它能够访问和处理的数据。OpenClaw提供了丰富的数据源集成能力，让Agent可以连接外部API、数据库、文件系统和消息队列，实现数据的实时获取、处理和流转。本章将详细介绍各种数据源的集成方法和最佳实践。

API数据源连接

OpenClaw支持通过HTTP/REST和GraphQL接口连接任意外部API。内置的API连接器提供了认证管理、请求限流、错误重试和响应缓存等功能。

# 在config.yaml中配置API数据源
data_sources:
  apis:
    weather_api:
      type: rest
      base_url: https://api.weather.com/v2
      auth:
        type: api_key
        header: X-API-Key
        key_env: WEATHER_API_KEY
      rate_limit: 100/min
      cache:
        enabled: true
        ttl: 300  # 缓存5分钟
      timeout: 10

    internal_api:
      type: graphql
      base_url: https://internal.example.com/graphql
      auth:
        type: oauth2
        token_url: https://auth.example.com/token
        client_id_env: INTERNAL_CLIENT_ID
        client_secret_env: INTERNAL_CLIENT_SECRET
      headers:
        Content-Type: application/json

在Agent中使用API数据源时，可以通过工具定义来声明需要的API：

# tools/weather_tool.py
from openclaw.tools import APITool

class WeatherTool(APITool):
    name = "get_weather"
    description = "获取指定城市的天气信息"
    data_source = "weather_api"

    async def execute(self, city: str, days: int = 1):
        response = await self.client.get(
            "/forecast",
            params={"city": city, "days": days}
        )
        return {
            "city": city,
            "forecast": response.data["forecast"],
            "temperature": response.data["temp"],
            "humidity": response.data["humidity"]
        }

提示：API认证安全

始终使用环境变量存储API密钥，切勿硬编码在配置文件中。OpenClaw支持多种认证方式：API Key、Bearer Token、OAuth 2.0、HMAC签名和自定义认证头。对于OAuth 2.0，平台会自动管理Token刷新。

数据库连接

OpenClaw支持连接主流的关系型数据库和NoSQL数据库，让Agent可以直接查询和操作数据。数据库连接通过连接池管理，确保高性能和资源安全。

关系型数据库：PostgreSQL、MySQL、SQLite、SQL Server
NoSQL数据库：MongoDB、Redis、Elasticsearch、DynamoDB
向量数据库：Qdrant、Pinecone、Weaviate、Milvus
时序数据库：InfluxDB、TimescaleDB

# 数据库数据源配置
data_sources:
  databases:
    main_db:
      type: postgresql
      host: ${DB_HOST}
      port: 5432
      database: production
      user: openclaw_reader
      password_env: DB_PASSWORD
      pool:
        min_size: 5
        max_size: 20
      ssl: true
      # 安全设置：只允许只读查询
      read_only: true
      query_timeout: 30
      allowed_schemas:
        - public
        - analytics

    cache_db:
      type: redis
      host: ${REDIS_HOST}
      port: 6379
      password_env: REDIS_PASSWORD
      db: 0

文件系统数据源

Agent可以通过文件系统连接器访问本地文件、网络存储和云存储中的文档。OpenClaw支持多种文件格式的自动解析，包括PDF、Word、Excel、CSV、Markdown和纯文本。

# 文件系统数据源配置
data_sources:
  file_systems:
    local_docs:
      type: local
      base_path: /data/documents
      allowed_extensions:
        - .pdf
        - .docx
        - .xlsx
        - .csv
        - .md
        - .txt
      max_file_size: 50MB
      watch_changes: true  # 监控文件变化

    cloud_storage:
      type: s3
      bucket: company-docs
      prefix: knowledge-base/
      region: cn-north-1
      credentials_env:
        access_key: S3_ACCESS_KEY
        secret_key: S3_SECRET_KEY

    # 文档解析和向量化管道
    pipeline:
      parsers:
        pdf: pytesseract  # OCR支持
        docx: python-docx
        xlsx: openpyxl
      chunking:
        strategy: semantic
        max_chunk_size: 1000
        overlap: 200
      embedding:
        model: text-embedding-3-small
        batch_size: 100

核心概念：文档处理管道

当Agent需要处理大量文档时，OpenClaw会自动启动文档处理管道。该管道包括文件解析、文本提取、分块（chunking）、向量化和索引存储等步骤。处理结果会缓存到向量数据库中，避免重复处理。

消息队列集成

对于需要实时数据流的场景，OpenClaw支持连接各种消息队列系统。Agent可以作为消费者订阅消息主题，也可以作为生产者发布消息。

# 消息队列数据源配置
data_sources:
  message_queues:
    kafka:
      type: kafka
      brokers:
        - kafka-1:9092
        - kafka-2:9092
        - kafka-3:9092
      consumer_group: openclaw-agents
      topics:
        - name: user-events
          handler: process_user_event
          auto_offset_reset: latest
        - name: system-alerts
          handler: process_alert
          auto_offset_reset: earliest

    rabbitmq:
      type: amqp
      host: ${RABBITMQ_HOST}
      port: 5672
      virtual_host: /openclaw
      queues:
        - name: task_queue
          prefetch: 10
          durable: true

数据管道与ETL

OpenClaw内置了轻量级的数据管道引擎，允许您定义从数据源到Agent的数据流转规则。这对于需要定期同步数据或进行数据转换的场景非常有用。

# 数据管道配置
pipelines:
  daily_report_pipeline:
    schedule: "0 8 * * *"  # 每天早上8点
    steps:
      - name: fetch_sales_data
        source: main_db
        query: |
          SELECT * FROM sales
          WHERE date = CURRENT_DATE - INTERVAL '1 day'
      - name: transform_data
        transform: python
        script: |
          def transform(data):
              total = sum(row['amount'] for row in data)
              return {
                  "date": data[0]['date'],
                  "total_sales": total,
                  "order_count": len(data),
                  "avg_order_value": total / len(data)
              }
      - name: send_to_agent
        target_agent: report_agent
        input_template: |
          昨日销售报告：
          日期：{{date}}
          总销售额：{{total_sales}}
          订单数量：{{order_count}}
          平均订单金额：{{avg_order_value}}

流式数据源

对于需要实时处理的数据流（如WebSocket、SSE或gRPC流），OpenClaw提供了流式数据源连接器。Agent可以持续接收数据更新，并实时做出响应。

WebSocket：适用于双向实时通信场景，如聊天应用、实时协作
Server-Sent Events (SSE)：适用于服务器向客户端推送更新，如股票行情、监控数据
gRPC Stream：适用于高性能的流式数据传输

警告：数据源安全

连接外部数据源时，请遵循最小权限原则。数据库连接应使用只读账户，API调用应限制请求频率，文件系统访问应限制在必要的目录范围内。所有数据源连接都应配置超时和重试策略，防止因外部服务不可用导致Agent阻塞。

通过灵活组合这些数据源集成能力，您可以构建出能够访问和处理各种数据的强大AI智能体。OpenClaw的数据源系统设计遵循插件化架构，您也可以轻松开发自定义的数据源连接器来满足特定需求。

Web 管理台

AI 模型管理