# InvesResearch Agent · 工程化产品需求文档(Engineering PRD) > 面向专业投资人 / 分析师的多市场(A股 / 港股 / 美股)智能体深度调研系统 > > **文档版本** v2.0(工程实施版) · **状态** 开发就绪草案 · **作者** 产品 + 架构 > **关联文档** 《系统设计文档 v1.0》《产品 PRD v1.0》《Harness 工程最佳实践调研报告》 > **本文档目标** 把产品构想推进到"工程可执行":定义系统契约、数据模型、Agent 状态机、降级策略、可测试的验收标准与 sprint 拆解,使工程团队可据此直接开工。 --- ## 0. 阅读指引与本文档的写作原则 这份 PRD 与上一版产品 PRD 的根本区别,在于它服务的是工程师而非利益相关方。产品 PRD 回答的是"我们为什么做、为谁做、做什么";这份工程 PRD 回答的是"怎么把它造出来、各部分之间的契约是什么、什么算做完了"。因此本文档遵循三条写作原则。 第一条原则是**契约先行**。在 agent 系统里,模糊性是最大的成本来源——一个没有明确输入输出 schema 的 Skill,会让编排层、前端、测试三方各自猜测,最终在集成时崩溃。所以本文档为每一个 Skill、每一个 MCP 工具、每一个 API 端点都给出明确的接口契约和 Pydantic 数据模型。 第二条原则是**验收即测试**。本文档里所有的功能需求都会附带 Given-When-Then 形式的验收标准,并且这些标准被设计成可以直接转化为离线 eval 数据集或自动化测试用例。一条"系统应该准确"这样的需求是无法验收的;而"给定贵州茅台的 5 年财务数据,DCF 工具在 WACC=8.5%、g=3% 假设下输出的每股内在价值应落在 1760–2140 区间,且所有现金流字段非空"是可以验收的。 第三条原则是**harness 与 agent 解耦**。这是 2025–2026 年 agent 工程领域最重要的范式转变,贯穿本文档的架构设计。我们把系统显式拆成三层:承载运行的工程骨架(harness)、可替换的模型与 agent 逻辑、以及双层能力封装(Skills 承载知识与流程,MCP 承载数据与计算)。这样做的理由是,大模型每升级一次,围绕它搭建的复杂度就要被重新定价——今天为了弥补模型短板而写的一堆 guardrail,明天可能就是纯粹的累赘。解耦的架构让我们能够在不动核心业务逻辑的前提下,既给 harness 做加法,也做减法。 --- ## 1. 系统概述与工程目标 ### 1.1 一句话定义 InvesResearch Agent 是一个把二级市场深度调研流程自动化的多智能体系统:用户给定一个标的或一个行业,系统编排一组协作的 AI agent 完成取数、财务诊断、估值建模、舆情分析、多空辩论与研报撰写,产出一份可作为正式研究起点的结构化报告。它的能力被封装成符合 Agent Skills 开放标准的能力包,既驱动自有的 Web 工作台,也能被 Claude Code、Codex、Gemini CLI 等任意兼容 runtime 直接调用。 ### 1.2 工程目标(可量化) 工程团队在本文档约定的范围内,需要交付一个满足以下可量化目标的系统。这些目标既是设计约束,也是最终验收的依据。 在性能上,单次标准深度调研(覆盖财务、估值、舆情、辩论、报告生成的完整流程)的端到端 P95 延迟应控制在 10 分钟以内,快速调研在 4 分钟以内;数据层在缓存命中时的响应应在 100 毫秒以内;系统应支持至少 20 个并发调研会话而不出现明显的延迟劣化。需要特别说明的是,我们用 P95 / P99 而非平均延迟作为指标,因为在 LLM 系统里平均延迟会掩盖掉那些让用户真正抓狂的长尾慢请求。 在成本上,单次完整深度调研的模型 token 成本硬上限是 0.5 美元,通过 model routing(强模型只用于规划、估值终审和投资结论,廉价模型承担检索、数据清洗、格式化、情绪打分)和分层缓存共同实现;系统须设置月度预算护栏,当单会话消耗触及预算的 80% 时告警,触及 100% 时熔断。 在可靠性上,单市场单数据源的数据可用率应大于 99%,这通过"统一适配器 + 优先级 failover + 熔断 + 退避重试"的组合实现;Agent 编排须采用 checkpoint 机制,任何一个节点失败都不应导致整个会话丢失,已完成的工作可恢复重跑。 在可观测性上,每一次 LLM 调用和工具调用都必须产生可追溯的 trace span,记录 token 归因(精确到哪个 agent、哪个用户、哪个功能消耗了多少)、延迟、成本和错误;eval 回归套件必须集成进 CI,任何会影响 agent 行为的变更在合并前都要跑过回归。 ### 1.3 非目标(明确排除,避免范围蔓延) 为了让工程团队聚焦,这里明确划出本系统不做的事。系统不提供个性化投资建议,所有产出附带显著免责声明,定位是研究效率工具;系统不做实盘交易和下单执行,不与券商交易系统对接;MVP 阶段不追求毫秒级 tick 数据,分钟级行情足够;系统不自建付费数据源的替代品,免费与开源数据的覆盖度和准确性边界就是产品能力的边界,这一点对用户透明披露。 --- ## 2. 架构原则与系统分层 ### 2.1 三层解耦:Harness / Agent Logic / Capability 整个系统的最高层抽象是三个解耦的层。理解这三层的边界,是理解后续所有设计决策的前提。 最底层是 **Harness(工程骨架)**,它是模型之外承载 agent 运行的一切:会话状态的持久化、编排循环、上下文管理、工具路由、沙箱执行、凭证边界、过程追踪和结果评估。Harness 的关键特性是它必须独立于具体的模型和 agent 逻辑演进——当我们把底层模型从 Claude Sonnet 换成 GPT-5,或者把某个 agent 的提示词改写,harness 层不应该需要改动。反过来,当模型能力增强让某些防护性的复杂度变得多余时(比如新模型不再陷入循环,那么循环检测中间件就成了纯成本),harness 应该能够把这部分逻辑干净地删除。 中间层是 **Agent Logic(模型与代理逻辑)**,它是可替换的业务大脑:每个 agent 的角色定义、提示词、辩论策略、模型选择。这一层会随着我们对投研流程理解的加深而频繁迭代,但它被 harness 托管,不直接触碰底层的状态持久化和工具调用机制。 最上层是 **Capability(能力封装)**,它采用 Skills + MCP 双层结构。Skills 承载"知识与流程"——比如 DCF 估值的标准作业流程、研报的写作规范、竞品对比的分析框架,这些是相对稳定、可以写下来在几周内都成立的领域知识。MCP 承载"数据与计算"——比如取财务数据、跑 DCF 计算、筛选股票、检索新闻,这些是每次调用结果都会变化、需要实时访问的能力。判定一个能力该放哪一层的最实用法则是:数据在调用之间会变就用 MCP,知识稳定到能写下来就用 Skill。 ### 2.2 Harness 运行时分层 把 Harness 这一层进一步展开,它由八个职责清晰的子系统组成。这套分层直接来自 2025–2026 年 agent harness 工程的业界共识,我们采纳它作为系统的骨架。 会话层(Session)负责持久化状态,由 LangGraph 的 Postgres checkpointer 实现,保证任何会话可中断、可恢复。Harness 控制平面负责编排循环本身——决定下一步执行哪个节点、如何恢复失败的步骤。上下文构建器(Context Builder)负责上下文的调度,包括接近窗口上限时的压缩(compaction)、工具结果的清理、子代理的上下文隔离。工具路由器(Tool Router)负责把 agent 的动作意图分发到正确的 MCP 工具。沙箱(Sandbox)负责隔离执行不可信代码。凭证代理(Credential Proxy)负责守住凭证边界——agent 永远看不到数据库密码或数据源 API key,只有 MCP server 持有这些凭证。追踪层(Trace)负责记录全过程,由 Langfuse 实现。评估层(Eval)负责对结果做评判。 这套分层的价值在于,每个子系统都有单一职责和清晰边界,可以独立测试、独立替换。当某个数据源失效时,问题被隔离在工具路由器和数据适配层,不会污染上下文管理或编排逻辑。 ### 2.3 技术栈决策与依据 技术栈的选择不是凭偏好,而是基于本系统的具体约束推导出来的。这里把每个关键决策和它的依据写清楚,以便团队理解"为什么是它"而非"为什么不是别的"。 后端语言选 **Python**,根本原因是金融数据生态。本系统赖以生存的数据库——AKShare、Tushare、baostock、yfinance、efinance、edgartools——全部是 Python 库;Pydantic 是所有主流 agent SDK 的数据校验层;LangGraph、Pydantic AI、FastMCP 都是 Python 一等公民。在这个领域用其他语言意味着要么自己重写数据适配层,要么跨语言调用,都得不偿失。 编排框架选 **LangGraph**。在与 CrewAI、AutoGen、OpenAI Agents SDK 的对比中,LangGraph 在生产可控性上最强:它提供显式的状态机(有向图 + 条件边)、内建的 checkpointing 与 durable execution、interrupt 机制支持人工介入、子图支持模块化、Send 原语支持并行 fan-out。独立基准显示 CrewAI 在简单工作流上的 token 开销约为 LangGraph 的三倍,而 AutoGen 的群聊模式每个任务要消耗 20 次以上的 LLM 调用,成本最高。对于一个既要多 agent 协作、又要严格控成本和保证可恢复性的系统,LangGraph 是当前最稳妥的选择。需要说明这个决策的可逆点:如果某个未来的迭代退化成纯粹的单 agent 调工具,可以降级用更轻的 OpenAI Agents SDK;如果需要超过五个角色的复杂协作且预算充足,可以重新评估已 GA 的 Microsoft Agent Framework。 结构化输出选 **Pydantic**。它是事实标准,所有主流 agent SDK 的校验层都用它。工程上我们优先用模型原生的 structured output(在生成层强制 schema,最可靠),其次是 tool/function calling,只在模型不支持原生约束时才退回到 prompted JSON。schema 设计要扁平,复杂数据拆成多步,并设置重试上限防止无限循环。 数据层的真相源选 **PostgreSQL / TimescaleDB**,分析层用 **DuckDB / Parquet**,热数据与限流用 **Redis**。TimescaleDB 在 Postgres 之上提供按时间自动分块的 hypertable 和约十倍的列式压缩,适合做系统真相和时序数据;DuckDB 是进程内的 OLAP 引擎,可以直接查询 Parquet 文件,适合做历史数据的分析扫描和回测,但它是单写者、durability 受限,所以只作只读分析层而非真相源;Redis 负责亚秒级的热报价缓存和限流令牌桶。 向量库的路径是 **pgvector 起步、Qdrant 上量**。在向量规模小于五千万、且已经有 PostgreSQL 的情况下,pgvector 是正确的默认选择,因为它不引入新的基础设施。当召回质量、复杂元数据过滤或规模成为瓶颈时,迁移到 Qdrant——它用 Rust 实现,过滤能力强,是金融这类需要复杂元数据过滤场景的首选。迁移的主要工作是重建索引,嵌入模型、分块策略和检索提示都可以平移。 可观测性选 **Langfuse 自托管 + OpenLLMetry 埋点**。Langfuse 是开源、OTel 原生、框架无关的,可以自托管甚至气隙部署——这对有数据不出域要求的机构客户至关重要。相比之下 LangSmith 虽然与 LangChain 生态最丝滑,但闭源、自托管仅限企业版且按 seat 加 trace 计费。用 OpenLLMetry 做埋点可以避免厂商锁定,必要时同时导出到多个 backend。 MCP server 用 **FastMCP** 实现,API 层用 **FastAPI**,前端用 **Next.js** 配合 SSE 做实时进度流。私有化部署为机构客户预留 **vLLM + 开源权重模型(Qwen / DeepSeek)** 的选项,vLLM 的 PagedAttention 相比朴素的 HuggingFace Transformers 有数量级的吞吐提升,适合自托管推理。 --- ## 3. 数据层设计(M1,最高优先级) 数据层是本系统成败的关键,也是工程风险最集中的地方,因此它是第一个里程碑,后续所有能力都依赖它。这里的核心挑战不是"如何调用某个数据库",而是"如何在一堆各有缺陷的免费数据源之上,搭建出一个对上层透明、稳定可靠的统一数据服务"。 ### 3.1 免费数据源的真实约束 在设计之前,工程团队必须清楚每个免费数据源的真实脾气,因为这些约束直接决定了路由和容灾策略。 AKShare 是非官方爬虫,聚合东方财富、新浪等源站的数据,覆盖最广(A股、港股、美股、期货、基金、宏观),不需要 API key,但它对上游网站的 HTML 结构和接口变化非常脆弱,典型故障是东方财富的限流导致连接被关闭。Tushare 用积分分级控制权限而非按次消耗,两千积分对应每分钟两百请求、每天十万请求的上限,港股美股日线需要五千以上积分,分钟数据是积分体系外的单独权限;它的官方建议是按交易日(一年约两百多天)而非按股票代码(五千多只)来批量拉取。yfinance 是非官方的 Yahoo Finance 封装,适合偶尔查询和小规模回测,但持续存在限流问题,会触发 IP 级封禁,且 VPN 也会被封,绝不适合连续大规模采集。baostock 免费免注册但只有 A 股,日线仅在夜间更新没有当日实时,且复权算法可能与通达信、同花顺不同需要交叉验证。efinance 覆盖 A 股但缺港股日线。edgartools 是 AI 原生的 SEC EDGAR 解析工具,提供干净的 API 来读取美股的 10-K、8-K、XBRL 财务、13F 等文件。 这些约束意味着:没有任何单一数据源能覆盖全部市场和全部数据类型,必须做组合;且每个源都可能在任意时刻失效,必须做容灾。 ### 3.2 统一适配器与市场感知路由 我们的设计借鉴两个成熟的参考实现:OpenBB Platform 的 provider abstraction(用统一 API 标准化多 provider,每个 provider 实现标准接口,共享字段用 Pydantic 元模型,在配置中声明优先级列表)和 daily_stock_analysis 项目的市场感知路由与熔断式冷却。 核心是一个 `DataAdapter` 抽象类,每个数据源实现一个具体的 `Provider` 子类。所有 provider 返回标准化的 Pydantic 模型(以 ticker + 数据频率作为唯一标识)。一个 `DataFetcherManager` 负责按市场和数据类型选择 provider,并维护优先级 failover。 ```python # 统一数据契约(简化示意,完整定义见第 4 节数据模型) from pydantic import BaseModel from enum import Enum from datetime import date class Market(str, Enum): A_SHARE = "SH_SZ" # 沪深 A 股 HK = "HK" # 港股 US = "US" # 美股 class DataType(str, Enum): QUOTE = "quote" # 实时报价 DAILY_BAR = "daily_bar" # 日线 FINANCIALS = "financials" # 财务三表 FILING = "filing" # 披露文件 NEWS = "news" # 新闻舆情 class Provider(ABC): """每个数据源实现此抽象。harness 不关心具体源,只面向此契约。""" name: str supported_markets: set[Market] supported_types: set[DataType] @abstractmethod async def fetch_daily_bar(self, ticker: str, start: date, end: date, adjust: AdjustType) -> list[DailyBar]: ... @abstractmethod async def health_check(self) -> bool: ... # 市场 × 数据类型 → 优先级 provider 列表(配置化,可热更新) ROUTING_TABLE: dict[tuple[Market, DataType], list[str]] = { (Market.A_SHARE, DataType.DAILY_BAR): ["tushare", "akshare", "baostock"], (Market.A_SHARE, DataType.FINANCIALS): ["tushare", "akshare"], (Market.HK, DataType.DAILY_BAR): ["akshare", "longbridge"], # HK 跳过 efinance/baostock (Market.US, DataType.DAILY_BAR): ["yfinance", "alphavantage"], (Market.US, DataType.FINANCIALS): ["edgartools", "yfinance"], (Market.US, DataType.FILING): ["edgartools"], # ... 其余组合 } ``` 路由的关键设计点是市场感知:港股请求绝不会被路由到只支持 A 股的 efinance 或 baostock,而是直接走 AKShare 或备用源。这避免了无谓的失败重试。 ### 3.3 容灾:熔断、退避、令牌桶 在路由之上,每个 provider 都被一个熔断器(circuit breaker)包裹。熔断器有三个状态:正常时闭合(CLOSED),连续失败达到阈值(比如五次)后打开(OPEN)并在冷却期内直接走 fallback 不再尝试该源,冷却期满后进入半开(HALF_OPEN)试探性恢复。这个设计直接借鉴 daily_stock_analysis 对 Longbridge 连接失败后的冷却处理,冷却时长可配置。 ```python class CircuitBreaker: def __init__(self, fail_threshold=5, cooldown_seconds=30): self.state = "CLOSED" self.fail_count = 0 self.opened_at = None # ... async def call(self, fn, *args, **kwargs): if self.state == "OPEN": if time.time() - self.opened_at < self.cooldown_seconds: raise CircuitOpenError # 触发上层 failover 到下一 provider self.state = "HALF_OPEN" try: result = await fn(*args, **kwargs) self._on_success() return result except RetriableError: self._on_failure() raise ``` 退避重试用 `tenacity`,采用指数退避加 jitter(随机抖动)以防止惊群效应,并严格区分可重试错误(429 限流、503 不可用、超时)和永久错误(404、认证失败)——后者不重试,直接 failover。限流用 Redis 实现的令牌桶,把每个数据源的速率上限(如 Tushare 每分钟两百次)中央化管理,跨进程共享。 ### 3.4 分层缓存与 TTL 策略 缓存是在免费数据源限频约束下保证响应速度的核心手段,采用四层结构。第一层是进程内的 LRU 缓存(用 cachetools),命中最快;第二层是 Redis,缓存热门报价、承载限流令牌桶,并用分布式锁防止缓存击穿(并发请求同一标的时塌缩成一次上游调用,这在限频环境下尤其关键);第三层是 DuckDB / Parquet,存储历史数据和基本面分析层;第四层是 PostgreSQL / TimescaleDB 作为系统真相源,只有在前面所有层都未命中时才触达免费 API。 TTL 策略按数据的更新节奏设定。实时报价 TTL 为 1 到 15 秒,只放在 Redis;日线数据在收盘后即定型不可变,TTL 设到下一个交易日,盘后刷新一次即可;季度财务数据 TTL 为数天到数周,在财报披露日失效;静态的公司元数据 TTL 为数周到数月,只在发生公司行动事件时刷新。 ### 3.5 数据层验收标准 数据层这个里程碑的完成标准是可测试的。给定 A 股市场的任意标的,在 Tushare 主源正常时,系统应在缓存命中时 100 毫秒内、未命中时 2 秒内返回标准化的日线数据,且所有字段符合 Pydantic schema;当主源被熔断时,系统应自动 failover 到 AKShare 并成功返回数据,整个过程对调用方透明;在连续一周的运行中,单市场单源的数据可用率应大于 99%,缓存命中率应大于 80%,且不应触发任何免费源的限流封禁。这些标准会被写成自动化测试和监控告警。 --- ## 4. 数据模型(Pydantic Schema 作为契约) 数据模型是整个系统的契约基础。我们用 Pydantic 模型定义所有跨模块流动的数据结构,它们既是运行时的校验器,也是文档,还是测试的基准。这里列出核心模型,完整定义随代码维护。 标的标识统一为 `MARKET:TICKER` 的形式。所有对外暴露的标的引用都通过一个 `Security` 模型规范化,内部一律使用这个统一 ID,避免 A 股的六位数字、港股的五位数字、美股的字母代码在系统里混用导致路由错误。 ```python class Security(BaseModel): uid: str # 统一 ID,如 "SH:600519" market: Market ticker: str # 原始代码,如 "600519" name_cn: str | None name_en: str | None currency: str # CNY / HKD / USD sector_gics: str | None sector_sw: str | None # 申万行业 class DailyBar(BaseModel): uid: str trade_date: date open: float; high: float; low: float; close: float volume: float adjust: AdjustType # 前复权 qfq / 后复权 hfq / 不复权 none source: str # 数据来源,用于追溯与交叉验证 class FinancialStatement(BaseModel): uid: str period: str # 报告期,如 "2025A" / "2025Q3" revenue: float | None gross_profit: float | None operating_income: float | None net_income: float | None operating_cash_flow: float | None total_assets: float | None total_equity: float | None # 关键指标做跨源校验,差异 > 10% 时打 data_quality_flag data_quality_flag: bool = False source: str class ValuationAssumptions(BaseModel): """DCF 假设。这是 human-in-the-loop 复核的对象。""" revenue_growth: float # 5 年营收增长率 net_margin: float wacc: float terminal_growth: float forecast_years: int = 10 class DCFResult(BaseModel): uid: str intrinsic_value_per_share: float value_range_low: float # 敏感性下限 value_range_high: float # 敏感性上限 current_price: float upside_pct: float assumptions: ValuationAssumptions fcf_by_year: list[float] # 分年自由现金流,供瀑布图 sensitivity_matrix: list[list[float]] # WACC × g class ResearchReport(BaseModel): report_id: str uid: str template: ReportTemplate # quick_note / deep_dive / sector / earnings_recap sections: list[ReportSection] rating: Rating | None # buy / hold / sell target_price: float | None target_price_range: tuple[float, float] | None version: int # 支持版本 diff created_at: datetime disclaimer: str # 强制非空,免责声明 ``` 注意 `FinancialStatement` 上的 `data_quality_flag` 字段:这是跨源校验机制的落点,当同一指标在不同数据源之间差异超过百分之十时,这个标志被置位,提醒上层和用户该数据存疑。`ResearchReport` 上的 `disclaimer` 字段被设为强制非空,从数据模型层面保证每份报告都带免责声明。 --- ## 5. 能力层:Skills 与 MCP 工具契约 能力层是 agent 实际调用的部分,分成 Skills(知识与流程)和 MCP 工具(数据与计算)两类。这一节为它们定义契约。 ### 5.1 MCP 工具契约 MCP server 用 FastMCP 实现,通过它暴露所有数据访问和计算能力。每个工具都有明确的输入输出 schema、是否破坏性的标记,以及错误码。凭证由 MCP server 持有,agent 永远看不到。 ```python from fastmcp import FastMCP mcp = FastMCP("investment-research") @mcp.tool() async def get_financials( uid: str, # 统一标的 ID,如 "SH:600519" period: str = "5y", # 周期:5y / 3y / ttm freq: str = "annual", # annual / quarterly ) -> list[FinancialStatement]: """获取公司财务三表(标准化)。覆盖 A股/港股/美股。 数据经统一适配层取得,带跨源校验标记。非破坏性,可缓存。""" ... @mcp.tool() async def run_dcf( uid: str, assumptions: ValuationAssumptions, ) -> DCFResult: """基于给定假设运行 DCF 估值,返回内在价值、敏感性矩阵、分年 FCF。 纯计算,确定性,非破坏性。假设的合理性由上层 HITL 复核。""" ... @mcp.tool() async def screen_stocks( universe: str, # 股票池,如 "hs300" / "sp500" factors: list[FactorFilter], # 因子条件 top_n: int = 20, ) -> list[ScreenResult]: """多因子选股。底层接 Qlib Alpha158/360 与 WorldQuant Alpha101。非破坏性。""" ... @mcp.tool() async def search_news( uid: str, lookback_days: int = 30, sources: list[str] | None = None, ) -> list[NewsItem]: """检索并返回标的相关新闻舆情。非破坏性。""" ... @mcp.tool() async def vector_search( query: str, collection: str, # filings / earnings_calls / news top_k: int = 5, ) -> list[RetrievedChunk]: """对研报/电话会/新闻向量库做语义检索(RAG)。非破坏性。""" ... ``` 工程上有几条强制约束。所有工具的输出都用 Pydantic 模型,启用 output schema;工具描述要清楚写明"做什么、何时用、是否破坏性、是否可缓存";破坏性工具(本系统 MVP 阶段几乎没有,但比如未来的"发送报告到邮箱")必须标记并触发 human-in-the-loop 确认;工具响应被当作数据而非指令处理,要剥离任何 ``、`` 之类的可疑标签以防 tool poisoning。 ### 5.2 Skill 契约 Skills 封装相对稳定的流程知识,符合 Agent Skills 开放标准。每个 Skill 是一个目录,顶层 `SKILL.md` 含 YAML frontmatter 和 Markdown 正文,辅以 scripts、references、assets 子目录。 下面是公司基本面调研 Skill 的契约示例。注意 frontmatter 的 `description` 字段同时承担"做什么"和"何时触发"两个职责,因为这是 progressive disclosure 第一级里 LLM 用来决定是否加载这个 Skill 的唯一依据。 ```markdown --- name: company-fundamental-research description: > 对单一上市公司(A股/港股/美股)进行基本面深度调研,产出业务模式、财务质量、 护城河、估值、风险点的结构化报告。当用户要求"调研某公司"、"看一下 600519/AAPL/00700"、"深度分析 XX 公司"或需要财务三表诊断时触发。 license: Apache-2.0 version: 1.0.0 allowed-tools: - investment-research:get_financials - investment-research:get_company_profile - investment-research:get_filings --- # Company Fundamental Research ## When to use 用户要求对一家上市公司做基本面调研时使用,支持三地市场。 ## Workflow 1. 标的标准化:调用 scripts/normalize.py 把输入归一化为 MARKET:TICKER。 2. 公司画像:get_company_profile,读业务分部、客户、地理收入分布。 3. 财务诊断:get_financials(period=5y),计算成长性、盈利能力、健康度; 遇到 data_quality_flag 为真的指标,在报告中标注存疑。 4. 关键披露:对最新年报调用 filing-parser 抽取 MD&A、风险、关联交易。 5. 同业对照:调用 peer-comparison 选 5-8 家做 PE/PB/PS 矩阵。 6. 产出:用 references/report_template.md 生成结构化 Markdown。 ## Output contract 产出必须符合 references/output_schema.json(对应 ResearchReport 模型)。 ``` Skill 的工程纪律包括:SKILL.md 正文控制在五千 token 以内(渐进披露第二级的预算),超出的细节放进 references 按需加载;name 必须与目录名匹配;MCP 工具在 Skill 内引用时用全限定名(如 `investment-research:get_financials`)以免多 server 共存时找不到工具;发布前必须跑过 skill 校验器,检查断链、名称不匹配、缺文件、token 超预算等常见错误。第三方 Skill 在引入前必须经过完整的源码审计,因为社区已出现过把恶意载荷藏在 Skill 工作流里的攻击案例。 ### 5.3 Skill 与 MCP 的边界划分 为了让团队在开发时不纠结,这里明确列出本系统每个能力归属哪一层。归在 Skill 的是流程知识:DCF 估值的标准作业流程、公司基本面调研 SOP、行业研究框架、研报写作规范、竞品对比方法论。归在 MCP 工具的是数据与计算:取财务数据、跑 DCF 计算、筛选股票、检索新闻、向量检索、取披露文件。判定依据始终是那条法则——结果每次调用都会变就是 MCP,知识写下来几周内都成立就是 Skill。 --- ## 6. Agent 编排:状态机与工作流 编排层用 LangGraph 把多个 agent 组织成一个可恢复、可观测、可人工介入的工作流。这一节定义状态机的结构。 ### 6.1 状态 Schema LangGraph 的状态 schema 是整个编排的中枢。设计原则是最小、显式、强类型,只在真正需要累积的字段上用 reducer。 ```python from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class ResearchState(TypedDict): # 输入 uid: str # 调研标的 depth: ResearchDepth # quick / standard / deep # 各 agent 产出(分析师并行写入) sector_analysis: SectorAnalysis | None company_analysis: CompanyAnalysis | None macro_analysis: MacroAnalysis | None financial_model: FinancialModel | None sentiment: SentimentResult | None valuation: DCFResult | None # 辩论过程(累积) debate_messages: Annotated[list, add_messages] debate_round: int bull_thesis: str | None bear_thesis: str | None neutral_verdict: str | None # 风控与结论 risk_assessment: RiskAssessment | None final_report: ResearchReport | None # 控制 needs_human_review: bool # 触发 interrupt 的标志 cost_tokens: int # 成本累计,触预算护栏 ``` 注意大部分字段是覆盖式的(每个 agent 写自己那块),只有 `debate_messages` 用了 `add_messages` reducer 做累积,因为辩论是逐轮叠加的。`cost_tokens` 字段在每个节点后更新,用于成本护栏检查。 ### 6.2 工作流图 工作流是一个有向图。Supervisor 节点解析用户意图,决定加载哪些 Skill 和走哪条路径(快速调研跳过辩论和深度估值)。分析师组通过 Send 原语并行 fan-out。估值在分析师完成后执行。多空辩论是一个带计数器的条件循环,Bull 和 Bear 交替发言,达到约定轮数后 Neutral 仲裁。风控官评估尾部风险。PM 综合所有输入产出结论。报告生成器固化为文档。 ```python from langgraph.graph import StateGraph, START, END from langgraph.types import Send, interrupt builder = StateGraph(ResearchState) builder.add_node("supervisor", supervisor_node) builder.add_node("sector_analyst", sector_node) builder.add_node("company_analyst", company_node) builder.add_node("macro_analyst", macro_node) builder.add_node("financial_modeler", modeler_node) builder.add_node("sentiment_analyst", sentiment_node) builder.add_node("valuation", valuation_node) builder.add_node("debate", debate_node) builder.add_node("risk_officer", risk_node) builder.add_node("pm", pm_node) builder.add_node("report_generator", report_node) # Supervisor 后并行 fan-out 到分析师组 def fan_out_analysts(state: ResearchState): targets = ["company_analyst", "sentiment_analyst"] if state["depth"] != ResearchDepth.QUICK: targets += ["sector_analyst", "macro_analyst", "financial_modeler"] return [Send(t, state) for t in targets] builder.add_conditional_edges("supervisor", fan_out_analysts) # 分析师 fan-in 到估值 for analyst in ["sector_analyst","company_analyst","macro_analyst", "financial_modeler","sentiment_analyst"]: builder.add_edge(analyst, "valuation") # 估值前的 human-in-the-loop:核心假设需人工复核 def valuation_node(state): assumptions = derive_assumptions(state) if state["depth"] == ResearchDepth.DEEP: # interrupt 暂停,等人工确认 WACC、永续增长率等假设 approved = interrupt({"assumptions": assumptions}) assumptions = approved or assumptions return {"valuation": run_dcf_with(assumptions)} # 辩论循环:达到轮数前回到 debate,之后进风控 def debate_router(state: ResearchState): if state["depth"] == ResearchDepth.QUICK: return "pm" if state["debate_round"] < MAX_DEBATE_ROUNDS: return "debate" return "risk_officer" builder.add_edge("valuation", "debate") builder.add_conditional_edges("debate", debate_router) builder.add_edge("risk_officer", "pm") builder.add_edge("pm", "report_generator") builder.add_edge("report_generator", END) # 生产用 Postgres checkpointer,支持中断恢复 graph = builder.compile(checkpointer=postgres_checkpointer) ``` ### 6.3 辩论模式 多空辩论是本系统信息密度最高的环节,直接借鉴 TradingAgents 的设计。Bull 研究员基于分析师产出构建看多论点,Bear 研究员构建看空论点,二者结构化地交锋至少两轮,每一轮都要针对对方的论点做出反驳而非各说各话。轮数(`MAX_DEBATE_ROUNDS`)和每个角色用的模型都是可配置的。Neutral 研究员在辩论结束后仲裁,识别双方分歧的核心(通常集中在某几个关键假设上),并据此给出基准情形和保守情形下的结论。这种纯自然语言的操作带来了相对传统量化方法的可解释性优势——用户能看到看多和看空的理由如何碰撞。 ### 6.4 成本与模型路由 成本控制内建在编排里。不同 agent 角色分配不同档位的模型:Supervisor 的意图解析、估值终审、PM 的投资结论、多空辩论用强模型(如 Claude Opus、GPT-5);情绪打分、新闻摘要、数据格式化、行业资料检索用廉价模型(如 Haiku、DeepSeek、Qwen)。这套路由通过 LiteLLM 或 OpenRouter 网关实现,网关同时提供 provider 级的 failover——当某个 LLM provider 宕机时自动切换。每个节点执行后更新 `cost_tokens`,当累计触及单会话预算的 80% 时记录告警,触及 100% 时编排提前终止并返回已完成的部分结果。 --- ## 7. 上下文工程 上下文是有限资源且边际收益递减,目标是用最小的高信号 token 集合驱动 agent。本系统采用四个手段管理上下文。 压缩(compaction)在上下文使用接近窗口上限时触发——参照 Claude Code 的经验,在估计达到窗口 90% 时压缩,75% 时先发警告。压缩采用两阶段 fallback:先把过大的工具返回截断到约五千字符,仍不够则对中间内容做摘要、保留头尾。 子代理隔离是最强的上下文防御手段。每个分析师 agent 在独立的上下文窗口里工作,只把结构化的摘要回传给编排层,而非把它探索过程中的全部原始数据塞回主上下文。这样做的好处是双重的:一个失败的子代理不会拖垮整个编排;一次消耗五万 token 的财务数据探索可以被压缩成两千 token 的诊断摘要回传。当某个操作预计会消耗超过上下文利用率的四到六成时,就应该主动派生子代理。 渐进披露由 Skills 机制天然实现:启动时只加载所有 Skill 的名称和描述(每个约一百 token),激活某个 Skill 时才加载它的完整正文,正文里引用的详细参考文档再按需加载。 外置与检索:历史调研、研报、电话会、新闻都存在向量库里,需要时通过 `vector_search` 动态检索,而非常驻上下文。 --- ## 8. 可观测性与评估 ### 8.1 可观测性 可观测性从第一天就要建立,用 Langfuse 自托管加 OpenLLMetry 埋点。每一次 LLM 调用和工具调用都产生一个 trace span。需要追踪的指标包括:token 归因(精确到哪个 agent、哪个用户、哪个功能消耗了多少),这对成本优化至关重要;延迟必须用 P95 和 P99 而非平均值,因为平均延迟会掩盖让用户抓狂的长尾;每次调用的成本;错误率;LLM-as-judge 的健康指标(不确定性、离散度、与人工标注的一致性);以及按数据切片(slice)的异常标记。 ### 8.2 评估驱动开发 评估是这份工程 PRD 区别于产品 PRD 的核心,因为它把"什么算做对了"变成可执行的检查。我们采用离线加在线的双层评估。 离线评估基于一个代表性的调研任务数据集,每个任务配有黄金答案或参考研报。这个数据集既用于建立基线指标,也用于回归测试——每当 prompt 或模型发生变更,就在同一数据集上重跑,对比指标看有无回归。评估方法分两种:确定性检查(JSON 能否正确反序列化、是否符合 Pydantic schema、引用覆盖率、数值是否落在合理区间)用于一切可以机械判定的地方,因为它们快速、便宜、可靠;LLM-as-judge(评判 faithfulness、相关性、推理质量)用于需要主观判断的地方,但必须配人工校准。 这里有一个必须警惕的工程陷阱:LLM judge 有系统性偏差,它对表层线索、文本长度、选项顺序敏感,而且 agent 的思维链不一定忠实反映其真实推理,可能只是事后的合理化。因此关键决策点绝不能只靠 LLM judge,要用确定性检查兜底,并定期用人工标注来校准 judge。 评估集成进 CI/CD:用 Braintrust 的 eval-action 或 Langfuse 的 dataset,在每个 PR 上自动跑评估并贴出每个 scorer 的改进或回归,任何会影响 agent 行为的变更在通过回归之前不得合并。 ### 8.3 验收标准的写法 本文档约定所有功能需求的验收标准都写成 Given-When-Then 的二元形式,覆盖正常路径、错误情况和边界情况,以便直接转化为测试用例。举两个例子。 公司基本面调研的一条验收标准:给定标的"SH:600519"且 Tushare 主源正常,当用户发起标准调研时,系统应在十分钟内产出一份符合 ResearchReport schema 的报告,报告包含非空的财务诊断、可比估值、投资评级和免责声明,且所有引用的财务数字都能追溯到 source 字段标注的数据源。 数据容灾的一条验收标准:给定 Tushare 主源连续五次请求失败,当系统再次请求 A 股日线数据时,熔断器应处于 OPEN 状态并自动 failover 到 AKShare,调用方应正常拿到数据且无感知,同时 Langfuse 中应记录一条熔断事件。 --- ## 9. 安全工程 MCP 和 Skills 是 2025–2026 年新增的主要攻击面,安全必须内建而非事后补。本系统遵循 OWASP MCP Top 10 做检查清单,重点防范几类威胁。 针对 prompt injection 和 tool poisoning(把恶意指令藏在工具描述或返回里),所有工具输出都被当作数据而非指令处理,在进入 agent 上下文前剥离 ``、`` 之类的可疑标签,对包含命令式语言的工具响应告警。针对供应链投毒(已出现过武器化的 Skill 和恶意 MCP 包,以及 CVE-2025-6514 这类客户端 RCE),所有第三方 Skill 和 MCP server 在上线前必须经过完整源码审计加 CI 扫描,本地 MCP server 在容器沙箱里隔离运行,依赖项做签名和 SBOM 跟踪。针对凭证泄露,严格执行凭证代理边界——agent 永远看不到数据库密码或数据源 API key,只有 MCP server 持有,且用短时、最小权限的 scoped 凭证。针对过度权限,每个 Skill 的 `allowed-tools` 做白名单,每个工具按最小权限授予。任何破坏性操作都要求人工批准。 合规层面,所有产出强制带免责声明(从 ResearchReport 数据模型层面保证),明确系统是研究效率工具而非投资建议;免费数据源的使用边界对用户透明披露,商业化前须取得授权或迁移付费源;为有数据不出域要求的机构客户提供 vLLM 私有模型加气隙自托管 Langfuse 的部署选项。 --- ## 10. MVP 工程拆解(里程碑与依赖) MVP 按六个里程碑推进,每个里程碑有明确的交付物、依赖关系和可测试的完成标准。里程碑之间是依赖链条,M1 是地基,后续都建立在它之上。 **M1 数据层地基。** 这是最高优先级,因为后续一切都依赖数据。交付统一数据适配器、市场感知路由、优先级 failover、熔断器、Redis 令牌桶限流、四层缓存,并打通 A 股单市场的 Tushare 主源加 AKShare 备源。完成标准是单市场单源数据可用率大于 99%、缓存命中率大于 80%、连续运行不触发限流封禁。这个里程碑不依赖任何其他里程碑。 **M2 单 Agent 加 MCP 工具。** 依赖 M1。交付 MCP server(FastMCP)、`get_financials` 和 `run_dcf` 两个核心工具、公司基本面调研 Skill、Pydantic 结构化输出,以及 Langfuse 埋点。完成标准是单个 agent 能对一个 A 股标的完成基本面调研并产出符合 schema 的结构化结果,全程可在 Langfuse 中追踪。 **M3 多 Agent 辩论编排。** 依赖 M2。交付 LangGraph 状态机、分析师组的并行 fan-out、多空辩论循环、风控官、PM 节点、Postgres checkpointer 以及估值假设的 human-in-the-loop interrupt。完成标准是深度调研能跑完完整的辩论流程,任意节点失败可从 checkpoint 恢复,人工可在估值假设处介入。 **M4 RAG 与舆情。** 依赖 M2。交付 pgvector 向量库、研报与新闻的 RAG 检索、`search_news` 与 `vector_search` 工具、情绪分析与量化筛选能力。完成标准是系统能基于历史研报和新闻做语义检索问答,情绪分析产出可追溯到来源。 **M5 报告生成与前端。** 依赖 M3、M4。交付研报生成 Skill(采用长任务 harness 的两段式——先由初始化 agent 写大纲和数据清单,再增量推进)、四种报告模板的 PDF / DOCX 导出,以及 Next.js 前端配 SSE 实时进度流。完成标准是用户能在 Web 上发起调研、实时看到各 agent 的执行进度、最终下载一份完整研报。 **M6 多市场扩展与私有化。** 依赖前述全部。交付港股与美股的市场感知路由(美股财务接 edgartools)、vLLM 私有部署选项、完整的 eval 回归套件和安全加固。完成标准是三地市场全覆盖、机构客户可选数据不出域的私有化部署、eval 回归集成进 CI。 --- ## 11. 待决策问题 进入开发前,有几个会影响架构和优先级的关键问题需要团队拍板。 第一是商业模式与定价,它直接影响成本控制的优先级和付费数据源的迁移时机——是按调研次数计费、按月订阅,还是开源加企业版双轨。第二是私有化部署的优先级,面向机构客户的数据不出域如果是硬性要求,会显著影响架构(模型是否必须支持本地部署、数据是否完全自托管),这关系到 M6 是否要提前。第三是免费数据源合规的法务边界,在多大范围、什么用途下使用这些爬虫聚合数据是安全的,需要法务给出明确意见,这是商业化的前提而非技术问题。第四是估值模块系统建议值的责任界定,如果系统给出的默认 DCF 假设导致用户决策失误,责任如何划分,这关系到产品的免责设计和功能边界。 --- ## 附录 A:架构决策记录(ADR)摘要 | 编号 | 决策 | 选择 | 主要依据 | 可逆点 | |---|---|---|---|---| | ADR-01 | 后端语言 | Python | 金融数据库与 agent SDK 生态全在 Python | 几乎不可逆 | | ADR-02 | 编排框架 | LangGraph | 生产可控、checkpointing、可观测最强 | 退化为单 agent 可降级 OpenAI Agents SDK | | ADR-03 | 结构化输出 | Pydantic + 原生 structured output | 事实标准,生成层强制 schema 最可靠 | 不可逆 | | ADR-04 | 真相源数据库 | PostgreSQL / TimescaleDB | 时序分块 + 列式压缩 + 连续聚合 | 可迁移 | | ADR-05 | 分析层 | DuckDB / Parquet | 进程内 OLAP,直查 Parquet,适合回测扫描 | 可替换 | | ADR-06 | 向量库 | pgvector → Qdrant | 小规模同栈起步,规模化迁 Rust 强过滤库 | 迁移成本=重建索引 | | ADR-07 | 可观测性 | Langfuse 自托管 + OpenLLMetry | 开源、OTel 原生、可气隙、避免锁定 | 可换 backend | | ADR-08 | MCP 框架 | FastMCP | 兼容最新 spec、Streamable HTTP、OAuth | 可替换 | | ADR-09 | 私有推理 | vLLM + Qwen/DeepSeek | PagedAttention 高吞吐,数据主权 | 仅机构场景启用 | | ADR-10 | 能力分层 | Skills(知识)+ MCP(数据计算) | 数据会变用 MCP,知识稳定用 Skill | 边界可调 | ## 附录 B:关键非功能指标汇总 | 维度 | 指标 | 目标值 | |---|---|---| | 性能 | 深度调研端到端 P95 延迟 | ≤ 10 分钟 | | 性能 | 快速调研端到端延迟 | ≤ 4 分钟 | | 性能 | 缓存命中响应 | ≤ 100 ms | | 性能 | 并发会话 | ≥ 20 | | 成本 | 单次深度调研 token 成本 | ≤ $0.5 | | 成本 | 预算护栏告警 / 熔断 | 80% / 100% | | 可靠性 | 单市场单源数据可用率 | > 99% | | 可靠性 | 缓存命中率 | > 80% | | 可观测 | trace 覆盖 | 100% LLM/工具调用 | | 质量 | Skill 调用成功率(MVP/差异化) | ≥ 90% / ≥ 95% | | 安全 | 第三方 Skill/MCP 上线前审计 | 100% | --- > **免责声明** — 本系统及其产出的所有内容仅用于研究效率提升与教育目的,不构成任何投资建议。投资有风险,决策需谨慎。系统输出受底层模型、温度、数据质量等非确定性因素影响,不保证准确性。本文档引用的开源项目数据与技术基准为 2025-2026 年快照,部分来自厂商博客与社区文章,实施前请核对官方文档。