# 多Agent协作的本地股票数据系统：从采集到应用的完整设计

在构建本地化股票数据系统的过程中，我经历了从单脚本到模块化再到多Agent协作的演进。最初一个大脚本包打天下，后来拆成采集、存储、分析三个模块。但随着数据维度增多、数据量增大，模块化方案也遇到了瓶颈：模块间耦合度高、数据流转不透明、扩展新数据源改动面大。

最终我选择了多Agent协作架构。每个Agent管理数据管道中的一个环节，Agent之间通过标准化消息协议通信，协作完成从原始数据到可用数据的全流程。这篇文章详细介绍这套多Agent架构的设计理念、核心组件和协作机制。

## 为什么选择多Agent架构？

股票数据管道有几个显著特点：一是数据异构性强，行情、财务、资金、L2指标的格式、更新频率、存储方式都不一样；二是环节依赖复杂，采集到清洗到存储到分析之间有大量状态传递和条件判断；三是需求变化快，经常需要接入新数据源、增加新分析维度。

微服务架构虽然能解耦，但引入了服务发现、分布式事务等复杂度，对个人或小团队过重。传统模块化灵活性不够，新增数据源往往需要修改多个模块的接口。

多Agent架构的优势在于：每个Agent是自主的智能体，具备感知、决策和行动能力，可以独立完成特定任务；Agent之间通过消息传递协作，松耦合且可扩展；每个Agent内部可以用最适合的技术栈实现。

## 架构总览：四个核心Agent

整个系统由四个核心Agent组成：采集Agent、清洗Agent、存储Agent、分析Agent。每个Agent负责数据管道的一个核心环节，协作完成从原始数据到可用分析结果的完整流程。另有一个调度中心负责协调Agent之间的工作。

```
                    ┌──────────────┐
                    │  调度中心    │
                    │ Orchestrator │
                    └──────┬───────┘
                           │ 消息调度
            ┌──────────────┼──────────────┐
            ▼              ▼              ▼
     ┌────────────┐  ┌────────────┐  ┌────────────┐
     │ 采集Agent  │  │ 清洗Agent  │  │ 存储Agent  │
     │ Collector  │──│  Cleaner   │──│  Storage   │
     └────────────┘  └────────────┘  └──────┬─────┘
                                             │
                                      ┌──────▼─────┐
                                      │ 分析Agent  │
                                      │  Analyzer  │
                                      └────────────┘
```

## Agent间的通信协议

Agent之间通过标准化消息协议通信。每条消息包含消息类型、发送方、接收方、关联ID、消息体、时间戳。

```python
from dataclasses import dataclass, field
from typing import Any, Optional
from enum import Enum
import uuid
from datetime import datetime

class MsgType(Enum):
    DATA_READY = "data_ready"
    DATA_REQUEST = "data_request"
    TASK_COMPLETE = "task_complete"
    ERROR = "error"

@dataclass
class AgentMessage:
    msg_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    msg_type: MsgType = MsgType.DATA_READY
    sender: str = ""
    receiver: str = ""
    correlation_id: Optional[str] = None
    payload: Any = None
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
```

消息传递采用发布-订阅模式。每个Agent可以订阅特定类型的消息，收到匹配消息时触发处理逻辑。

```python
class MessageBus:
    def __init__(self):
        self.subscribers = {}
        self.agents = {}

    def subscribe(self, agent_id, msg_types):
        for msg_type in msg_types:
            self.subscribers.setdefault(msg_type, []).append(agent_id)

    def publish(self, message):
        receivers = self.subscribers.get(message.msg_type, [])
        for agent_id in receivers:
            if agent_id == message.receiver or message.receiver == "*":
                self.agents[agent_id].on_message(message)
```

## 采集Agent：数据入口的守护者

采集Agent是整个数据管道的起点，负责从各个数据源获取原始数据。它管理多个数据接口、处理并发调用、控制请求频率、处理网络异常。

内部架构分为三个部分：接口管理器、任务调度器、异常处理器。

接口管理器维护所有可用的数据接口，包括接口地址、参数模板、调用频率限制等。每个接口注册成一个可调用的服务。

```python
class CollectorAgent:
    INTERFACES = {
        "stock_list": {"path": "/base/gplist", "rate": 1},
        "realtime": {"path": "/time/real/{dm}", "rate": 5},
        "kline": {"path": "/time/history/trade/{dm}/{level}", "rate": 3},
        "finance": {"path": "/time/f10/fi/{dm}", "rate": 2},
        "l2_sign": {"path": "/time/real/trace/l2sign/{dm}", "rate": 5},
        "onebyone": {"path": "/time/real/trace/onebyone/{dm}", "rate": 5},
        "zlzjzs": {"path": "/time/zijin/zlzjzs/{dm}", "rate": 3},
        "zjlrqs": {"path": "/time/zijin/zjlrqs/{dm}", "rate": 3},
        "longhubang": {"path": "/time/data/longhubang", "rate": 1},
        "bshgt": {"path": "/time/data/bshgt", "rate": 1},
    }

    def collect(self, task):
        interface = self.INTERFACES[task["interface"]]
        results = []
        rate_limiter = RateLimiter(interface["rate"])
        for dm in task["dm_list"]:
            rate_limiter.acquire()
            url = self._build_url(interface["path"], dm)
            result = self._call(url)
            if result.success:
                results.append(result)
            else:
                self._handle_error(dm, task["interface"], result)
        self._publish_results(task["interface"], results)
```

任务调度器支持定时触发和事件触发。定时触发由调度中心发出，比如交易日15:30触发全市场K线更新；事件触发由其他Agent的数据请求消息触发。

异常处理器管理接口超时、返回格式错误、连续失败等异常，每种异常有对应处理策略，从简单重试到切换备用接口。

## 清洗Agent：数据质量的第一道防线

清洗Agent接收采集Agent传来的原始数据，进行格式标准化、异常值过滤、字段映射、数据校验，确保合格数据才能进入存储环节。

核心是一套规则引擎。每条清洗规则定义了输入条件、处理动作和输出结果。规则可以灵活配置，支持条件组合和优先级排序。

```python
class CleanerAgent:
    FIELD_MAPPING = {
        "f43": "cjjg", "f47": "cjl", "f58": "cjsj", "f169": "jyzd",
        "f62": "zlJlr", "f184": "zlJlb", "f70": "shJlb",
        "f164": "ddx", "f165": "ddy", "f166": "ddz", "f167": "ddf",
        "f78": "f5MinZlJe",
    }

    VALIDATION_RULES = [
        {"field": "dm", "check": lambda v: len(str(v)) == 6, "on_fail": "filter"},
        {"field": "cjjg", "check": lambda v: v is None or v > 0, "on_fail": "filter"},
        {"field": "cjl", "check": lambda v: v is None or v >= 0, "on_fail": "filter"},
        {"field": "zdl", "check": lambda v: v is None or abs(v) <= 20, "on_fail": "warn"},
    ]

    def on_message(self, message):
        if message.msg_type == MsgType.DATA_READY:
            self._process_batch(message.payload)

    def _process_batch(self, payload):
        records = payload["records"]
        cleaned, filtered, warnings = [], 0, 0
        for record in records:
            mapped = self._apply_mapping(record.raw_data)
            result = self._validate(mapped)
            if result.cleaned:
                cleaned.append(result.cleaned)
            filtered += result.filtered
            warnings += result.warnings
        self._publish(cleaned, payload["batch_id"], filtered, warnings)
```

清洗过程中被过滤的送到异常数据仓库，生成清洗报告记录每批数据的输入量、输出量、过滤原因。

## 存储Agent：数据资产的管理者

存储Agent接收清洗后的数据，持久化到对应的存储引擎。它管理多种引擎，根据数据类型选择最合适的方案。

```python
class StorageAgent:
    ENGINE_ROUTING = {
        "stock_list": "sqlite",
        "realtime": "redis",
        "kline": "duckdb",
        "finance": "duckdb",
        "zlzjzs": "duckdb",
        "l2_sign": "duckdb",
        "onebyone": "duckdb",
    }

    ENGINES = {
        "sqlite": SQLiteEngine("stock.db"),
        "duckdb": DuckDBEngine("stock.duckdb"),
        "redis": RedisEngine(),
        "json": JSONEngine("config/"),
    }

    def on_message(self, message):
        if message.msg_type == MsgType.DATA_READY:
            engine_name = self.ENGINE_ROUTING.get(message.payload["source"], "sqlite")
            engine = self.ENGINES[engine_name]
            version = datetime.now().strftime("%Y%m%d_%H%M%S")
            count = engine.write(message.payload["records"], version)
            self._report_success(message.payload["batch_id"], count, version)
        elif message.msg_type == MsgType.DATA_REQUEST:
            data = engine.query(message.payload)
            self._deliver(message.msg_id, data)
```

引擎路由根据数据类型选择存储：基础信息用SQLite，K线用DuckDB，实时行情用Redis，配置用JSON。版本控制确保每次写入都有版本号，支持数据回溯。

## 分析Agent：数据价值的创造者

分析Agent从存储Agent获取数据，执行分析任务，生成结果。它的核心是任务队列和多个分析模块。

```python
class AnalyzerAgent:
    MODULES = {
        "basic_stats": BasicStatsModule(),
        "technical": TechnicalIndicatorsModule(),
        "factor": FactorModule(),
        "backtest": BacktestModule(),
        "correlation": CorrelationModule(),
    }

    def submit_task(self, task):
        self.task_queue.append(task)
        self.task_queue.sort(key=lambda t: t.priority)
        self._process_next()

    def _process_next(self):
        if not self.task_queue:
            return
        task = self.task_queue.pop(0)
        data = self._request_data(task.requirements)
        module = self.MODULES[task.module]
        result = module.execute(data, task.params)
        self._deliver_result(task, result)
```

分析模块覆盖基础统计、技术指标、因子计算、策略回测、相关性分析等场景。

## 协作流程：一次完整的数据旅程

以"全市场日K线更新"为例，展示多Agent的协作流程。

调度中心在交易日15:30触发定时任务，向采集Agent发送采集指令。采集Agent获取全市场股票列表，分批调用K线接口，每批数据发送给清洗Agent。清洗Agent执行字段映射和数据校验，合格数据发送给存储Agent。存储Agent选择DuckDB引擎写入，生成版本号，完成后报告调度中心。调度中心触发分析Agent计算日涨跌幅、成交量变化等指标，生成市场概览报告。

整个流程中每个Agent只关注自己的任务，通过消息传递协作。某个环节失败不会阻塞其他环节。

```python
DAILY_PIPELINE = [
    {"trigger": "15:30", "steps": [
        {"agent": "collector", "action": "collect_kline"},
        {"agent": "collector", "action": "collect_realtime"},
        {"agent": "collector", "action": "collect_finance"},
        {"agent": "analyzer", "action": "generate_daily_summary"},
        {"agent": "analyzer", "action": "update_factor_database"},
    ]}
]
```

## 容错与演进

多Agent架构的关键优势是容错能力。每个Agent有独立的健康监控，定期发送心跳。调度中心连续收不到心跳会触发恢复流程：重启Agent进程、恢复未完成任务、重新派发失败消息。

降级策略方面，主数据源不可用时采集Agent自动切换备用接口，主存储引擎不可用时存储Agent切换备用引擎。

这套架构运行8个月来，从最初3个Agent扩展到4个，经历了数次调整。体会最深的有三点：Agent的边界划分比技术实现更重要，采集关注外部接口稳定性，清洗关注内部数据质量，拆分后代码更简洁；消息协议设计要向前兼容，避免升级时的痛苦；不要过度设计，先实现核心Agent，在使用中迭代扩展。

未来计划引入Agent自主决策能力，让分析Agent根据市场状态自动选择分析模型；建立Agent间的学习机制，共享知识和经验；引入对话式交互，让用户用自然语言请求数据分析任务。

资料参考：ig50

在构建本地化股票数据系统的过程中，我经历了从单脚本到模块化再到多Agent协作的演进。最初一个大脚本包打天下，后来拆成采集、存储、分析三个模块。但随着数据维度增多、数据量增大，模块化方案也遇到了瓶颈：模块间耦合度高、数据流转不透明、扩展新数据源改动面大。

多Agent协作的本地股票数据系统：从采集到应用的完整设计

数据库

后端

架构设计

多Agent协作的本地股票数据系统采用采集、清洗、存储、分析四大智能体，通过标准化消息协议实现松耦合数据管道。系统支持异构数据源接入、灵活扩展新数据源，具备容错降级机制，实现从原始数据到分析结果的自动化流转，适合个人及小团队本地化部署。

微服务架构

数据接口

数据仓库

数据分析

数据管道

代码解释

数据流

智能体

路由

SQLite

Redis

Agent

JSON

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

多Agent协作的本地股票数据系统：从采集到应用的完整设计-腾讯云开发者社区-腾讯云

多Agent协作的本地股票数据系统：从采集到应用的完整设计

多Agent协作的本地股票数据系统：从采集到应用的完整设计

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐