企业官网建设流程全解析

做投资研究这几年，我最大的感受就是信息量太大，一个人根本看不过来。所以我从去年开始尝试把AI Agent引入投资研究工作流，用多智能体分工处理数据采集、新闻舆情、财报分析和风险总结，把原来每天要花两三个小时的盯盘和翻资料，压缩成十分钟就能拿到一份带完整逻辑链和来源引用的结构化报告。这篇文章就把我的经验整体整理一遍，从架构设计、技术选型到关键代码、踩坑记录都会讲到，最后还有一套给新手的从0到1练手路径。适合正在做量化投研、准备用Agent替代重复劳动的朋友参考。

1. 为什么选择用AI Agent重构投资研究流程

1.1 传统投资研究方式的真实痛点

先说清楚我要解决的痛点。做股票、期货这类二级市场的投研，本质上是在跟信息赛跑，而信息是无限膨胀的。以前我的流程大致是：早晨起来先看隔夜外盘，再刷一遍财经新闻，然后翻公告、研报，最后自己手工汇总成当天的观察清单。这套流程有两个致命的毛病：

第一是覆盖不全。一个人精力的上限摆在那里，盯了三五个行业，其他行业冒出的重要变化根本没精力处理。我试过同时跟踪二十只股票，每只股票每天光公告、新闻、大宗交易记录就能产生几十条信息，这还没算技术面的价格形态。坚持了两周就发现，所谓跟踪基本变成了“看标题式关心”，很多关键信息其实都漏过去了。

第二是情绪干扰。人天生容易受账户浮盈浮亏影响，涨了容易乐观，跌了容易悲观。这导致同一个消息，在不同情绪状态下，你解读出来的“风险等级”完全不一样。更麻烦的是，这种偏差你自己很难察觉。后来我意识到，如果能把信息收集、初步分析、风险提示这些重复劳动交给程序，自己只做决策层的工作，就能大幅减少这种情绪污染。

1.2 我期待Agent方案解决什么问题

开始动手前，我给自己列了几个明确的目标：信息采集要全，分析过程要快，结论要能回溯。所谓“回溯”，就是说任何一个结论，比如“这只股票最近负面舆情增多”，你点开之后要能看到它依据了哪几条新闻、哪几份公告，而不是黑盒模型告诉你一个结论就没有然后了。

这也是我最终放弃“直接调用大模型API问几个问题”这种简单方案的原因。单次问答模式看起来能用，但做不到持续跟踪和多人协作，更做不到按标准流程批量处理几十只股票。AI Agent恰恰适合这种场景——它可以把“搜集数据—分析基本面—评估新闻情绪—生成报告”拆成多个步骤，每一步都有明确输入输出，还能动态调整。相当于我请了四个研究员，各自守着一块任务，最后把分析结果汇总到我这里。

1.3 Agent方案和RPA、普通脚本的本质区别

有人可能会说，这不就是爬虫加脚本吗？我之前也做过纯爬虫方案，写死了采集频率和规则，结果每次业务逻辑一调整就要改一堆代码。Agent方案最大的不同在于它有推理和规划能力。

还是用新闻分析举例子。RPA脚本能做的极限是：抓取新闻列表，按关键词过滤，然后输出一个“提到XX公司多少条”的统计。但AI Agent能做的是：先识别出新闻里涉及的公司、事件类型、影响方向，再根据信源权威性加权，最后结合当前股价位置给出“这个消息已经被price in多少”的推测。这些步骤RPA没法用固定规则写死，因为每一条新闻的表达方式都不同，只有具备语义理解能力的模型才能在一个动态流程里完成。

2. 整体架构设计与核心流程

2.1 系统整体架构分层

整个项目我分成了四层：数据接入层、分析Agent层、流程编排层、服务输出层。每一层的职责都很明确，没有把逻辑混在一起。

这套分层的出发点是可替换性。比如今天AkShare接口改版，我只需要改数据接入层里的一个小模块，分析Agent和编排层完全不用动。同样的道理，今天想从GPT换到DeepSeek或Qwen，只需要在模型工厂里改一个配置项，所有Agent的代码框架保持不变。

2.2 数据源选型与接入策略

数据源是整个项目的地基，这里踩的坑最多。我的选型逻辑很简单：能用免费稳定接口解决的不用爬虫，能一次拿全量数据的不要逐条请求。

行情数据我主要用AkShare和Tushare两个库。AkShare胜在接口全、更新快，从A股实时行情到期货夜盘数据都有覆盖；Tushare胜在数据结构规范、权限体系完整，做历史回测时更可靠。我的实测经验是：日常实时行情用AkShare就够，但如果要拉五年以上的日线数据做回测，Tushare的积分接口更稳。

新闻类数据是最麻烦的。财经网站有反爬机制，而且不同网站的栏目结构经常变。我的做法是自建一个小型采集服务，定时抓取几个可信度比较高的财经门户的财经头条和个股新闻板块，存到数据库后做去重和清洗。这里特别提醒一句：抓取公开网页要注意目标网站的robots协议和访问频率，别把别人的站点打挂了。我这边控制单IP请求频率在每秒两次以内，同时设置请求失败自动退避。

公告数据我用巨潮资讯网的公开接口，这个源的好处是权威、格式统一。财务数据则在财报季从AkShare对准报告期批量拉取。宏观数据如果不涉及敏感细分项，也可以用公开的统计接口，但我不做预测只做参考，这点后面会细说。

2.3 分析Agent如何分工

我设计了五个Agent，每只只负责一件事，这样提示词可以写得非常聚焦，不会出现一个Agent什么都会但什么都做不精的情况。

• 新闻舆情Agent：负责抓取目标标的相关新闻，做情感极性判断和热度趋势分析。
• 基本面Agent：负责读财报指标，比如营收增速、净利率、资产负债率，判断与上一报告期的变化情况。
• 技术面Agent：负责计算均线、MACD、量价配合度，生成短期趋势描述。
• 估值Agent：负责把当前市盈率、市净率放到历史区间里，输出“处于历史什么水位”的判断。
• 综合风控Agent：负责汇总前面四个Agent的输出，检查是否有矛盾点，生成最终的风险提示和总结报告。

这种分工还有一个好处，就是单Agent的上下文窗口占用非常低，不至于把新闻全文一股脑塞给一个模型。金融大模型本来就容易在长上下文里丢失关键信息，切短输入以后准确率提升很明显。

3. 关键实现与实操记录

3.1 环境准备与项目结构

整个项目基于FastAPI和LangGraph构建，模型层用LangChain的OpenAI兼容接口，方便随时切换不同厂商的模型。如果你要复现，环境依赖大概是这样：

pip install fastapi uvicorn langchain langchain-openai langgraph pip install akshare tushare pandas redis asyncpg

这些依赖不代表你全部都要用，AkShare的数据如果只走同步接口，不装asyncpg也没关系。我的原则是，先让流程跑通，再考虑性能和并发，不要在第一步就引入过多依赖。

项目目录我按模块组织，入口很清晰：

investment_research/ ├── main.py # FastAPI入口 ├── agents/ # 各Agent的提示词和调用逻辑 │ ├── news_agent.py │ ├── fundamental_agent.py │ ├── technical_agent.py │ └── risk_agent.py ├── datasources/ # 数据接入层 │ ├── market.py │ └── news.py ├── workflows/ # LangGraph编排 │ └── research_graph.py └── services/ # 业务服务层 └── research_service.py

别看目录小，我后来加了定时任务模块、缓存模块、回调告警模块，目录结构基本还能维持住，靠的就是一开始把“数据”和“分析逻辑”彻底分离。

3.2 数据接入模块实现

先拿行情数据举例。AkShare的实时行情接口返回的是全市场快照，数据量很大，我每次只提取目标代码那一行：

import akshare as ak import pandas as pd def fetch_realtime_quote(symbol: str) -> dict: df = ak.stock_zh_a_spot_em() row = df[df["代码"] == symbol] if row.empty: return {"error": f"symbol {symbol} not found"} row = row.iloc[0] return { "symbol": symbol, "name": row["名称"], "price": float(row["最新价"]), "change_pct": float(row["涨跌幅"]), "volume": float(row["成交量"]), "amount": float(row["成交额"]), }

这个接口实测第一次调用会比较慢，因为要拉全市场数据，所以我加了缓存——同一分钟内请求同一只股票，直接返回上一次结果，避免反复全表扫描。新闻采集也类似，抓下来的标题和正文会存入本地SQLite，再做一次简单的去重，按股票代码建立索引。

3.3 新闻舆情Agent的构建

舆情Agent是整个系统里最有用的一个，也是提示词写得最细致的。我给它设计了三个输出维度：情感得分、影响领域、热度趋势。

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate model = ChatOpenAI( model="qwen-plus", temperature=0.2, max_tokens=512, ) prompt = ChatPromptTemplate.from_messages([ ("system", """ 你是专业的金融舆情分析师。请根据下面的新闻标题和摘要，分析这条新闻对指定上市公司的影响。 要求： 1. 先判断情感倾向，输出positive/neutral/negative之一。 2. 给出一个-1到1之间的情感分数，正数代表利好。 3. 判断影响领域，只能从【业绩】【管理层】【行业政策】【市场情绪】【供应链】中选一个最匹配的。 4. 用不超过50字说明判断理由，必须引用新闻原文中的关键句。 """), ("human", "公司：{company}\n新闻时间：{time}\n新闻标题：{title}\n新闻摘要：{summary}"), ]) def analyze_one_news(news_item: dict) -> dict: result = model.invoke({ "company": news_item["company"], "time": news_item["time"], "title": news_item["title"], "summary": news_item["summary"], }) # 这里会接一个JSON解析器，把结果转成结构化字段 return parse_llm_output(result.content)

核心点在于“必须引用新闻原文中的关键句”。加了这个约束之后，模型编造理由的情况少了很多，就算它判断错了，我也能顺着它引用的句子去复核，不至于出现完全找不到依据的结论。

3.4 用LangGraph编排多Agent流程

LangGraph对我来说最大的价值是状态传递和条件分支。它不像LangChain原来的Chain那样固定顺序，而是能根据前的节点输出决定下一步走向，这个特性在投研场景里太实用了。

举个实际例子：基本面Agent如果发现最新财报季数据还没披露，就跳到“等待季报模板”分支，不再去做无意义的盈利预测；如果新闻舆情Agent发现负面新闻数量超过阈值，风控Agent就会被提前触发，在最终报告里加大风险警告权重。

from langgraph.graph import StateGraph, END from typing import TypedDict, List class ResearchState(TypedDict): symbol: str news: List[dict] fundamental: dict technical: dict risk_score: float report: str def collect_data(state: ResearchState) -> ResearchState: # 拉取行情和新闻，这里做缓存处理 state["news"] = news_service.get_related(state["symbol"]) return state def analyze_fundamental(state: ResearchState) -> ResearchState: state["fundamental"] = fundamental_agent.run(state["symbol"]) return state def analyze_technical(state: ResearchState) -> ResearchState: state["technical"] = technical_agent.run(state["symbol"]) return state def analyze_news_sentiment(state: ResearchState) -> ResearchState: neg_count = sum(1 for n in state["news"] if n["sentiment"] == "negative") state["risk_score"] = state.get("risk_score", 0) + neg_count * 0.1 return state def generate_report(state: ResearchState) -> ResearchState: state["report"] = risk_agent.summarize(state) return state graph = StateGraph(ResearchState) graph.add_node("collect", collect_data) graph.add_node("fundamental", analyze_fundamental) graph.add_node("technical", analyze_technical) graph.add_node("news", analyze_news_sentiment) graph.add_node("report", generate_report) graph.set_entry_point("collect") graph.add_edge("collect", "fundamental") graph.add_edge("collect", "technical") graph.add_edge("collect", "news") graph.add_edge("fundamental", "report") graph.add_edge("technical", "report") graph.add_edge("news", "report") graph.add_edge("report", END) app = graph.compile()

compile之后就可以直接传初始状态运行了。LangGraph还会自动把每一步的状态变化记录下来，这对事后复盘特别重要——我可以清楚看到一条结论是在哪一步、基于什么数据产生的。

3.5 FastAPI接口与并发扛量实践

热搜里有个词很准：“AI Agent怎么扛并发”。我实际开发中确实被并发问题教育过。一开始我只是简单把流程串起来，单线程跑，然后发现两个问题：一是同步请求大模型太慢，一个标的完整流程要一二十秒；二是多个请求同时进来，程序直接卡死。

我的解决方案分几层。首先是FastAPI接口把同步函数改成异步，并且用后台任务模式处理：“接收请求—立即返回任务ID—后台跑研究流程—完成后主动推送给用户”。这样用户体验好很多，不用干等。

import asyncio from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel app = FastAPI() sem = asyncio.Semaphore(10) # 控制同时执行的Agent任务数 class ResearchRequest(BaseModel): symbol: str @app.post("/research") async def create_research(req: ResearchRequest, background_tasks: BackgroundTasks): task_id = generate_task_id(req.symbol) background_tasks.add_task(run_research_with_semaphore, task_id, req.symbol) return {"task_id": task_id, "status": "queued"} async def run_research_with_semaphore(task_id: str, symbol: str): async with sem: await run_research_pipeline(task_id, symbol)

这里最难控制的是大模型的QPS配额。不同厂商的模型都有速率限制，一旦超限就会被限流甚至封禁。我的做法是维护一个信号量，限制同时进行的大模型调用数，另外加了一层带指数退避的重试机制。如果有连续失败，就把任务丢进延迟队列，避免雪崩。

除了接口层，Redis缓存也很关键。我做了两级缓存：第一级是对数据源的缓存，AkShare全市场行情两分钟内只拉一次；第二级是对分析结果的缓存，同一只股票在股票池里短时间重复请求，直接用上次生成的报告，不再重复调用大模型。实测下来，并发从5路撑到30路没有太大问题，成本也省了一半多。

4. 踩坑与排查实录

4.1 大模型幻觉，金融场景尤其严重

不管用什么模型，投资研究里最大的风险就是幻觉。金融数据讲究精确到小数点，模型如果编一个假的毛利率出来，那整个分析就废了。

我踩过最离谱的一次，是让基本面Agent总结某公司的盈利情况，模型信誓旦旦地写“净利润同比大幅增长35%”，我拿原始财报一核对，实际数据是下降12%。后来我加了两个防线：一是所有数字必须带出处，模型必须给出它引用的公告编号或字段名称；二是增加一个独立的校验Agent，专门去对比模型输出和结构化财务数据，不一致就标记为“数据冲突”并重新生成。加了这两道防线之后，基本再没出现过明显的数字幻觉。

4.2 上下文塞爆与Token超限

新闻数量一多，把所有正文塞进Prompt肯定不行。有一次我跟踪一只热门股票，当天相关新闻六十多条，全塞进去直接超过了模型的上下文窗口，跑出一个报错。

这个问题我用MapReduce的方式解决。先让模型对每条新闻只抽“情感标签+影响领域+关键句”，不写长分析；然后汇总所有标签再做整体情绪判断。这样每条新闻最多消耗两三百个Token，六十条也才一万多Token，完全在可控范围。再往后新闻量更大时，我用向量数据库做过一版，先把新闻嵌入存储，分析时只取跟当前标的相关度最高的二十条，效果更稳。

4.3 Agent跑飞、死循环和超时

多Agent编排最担心的就是流程跑飞。LangGraph虽然提供了状态机能力，但节点之间如果某个工具返回了意外格式，节点函数没有做好容错，整个流程就会卡在中间节点不回传。

我遇到过一个典型问题：新闻采集服务被目标网站反爬拦截，返回了空列表。新闻Agent拿到空列表后继续往下一个节点传，风控Agent发现“没有新闻”，竟然给出“舆情平稳”的结论——这等于把数据缺失误判成利好，很危险。修复方案是每个Agent节点加校验：输入数据不满足最小数量时，必须在结论里标注“数据缺失，可信度降级”，并且不允许直接把缺失当正常。

另外一定要设置LangGraph的recursion_limit和每个节点的执行超时。我最初没设超时，有一次模型服务超限导致节点重试了二十多次，白白浪费了很长时间。后来统一设定为单节点最多重试3次、单次等待不超过30秒，跑飞的情况基本绝迹。

4.4 成本控制：单次分析烧掉的Token比想象中多

投资研究的Agent流程会反复调用大模型，如果设计不当，成本会高得吓人。我第一版跑完整流程时，单只股票的分析居然要消耗五万Token，一个月跟踪二十只股票，费用直接超预算。

优化思路主要有三条。第一条，能用小模型解决的绝不用大模型：新闻情感分类这类简单任务用7B量级的模型就很好，综合报告这种需要深度推理的才上最强的模型。第二条，多Agent之间传递中间结果时，只传结构化摘要，不传原始文本，减少Token量。第三条，相似的查询请求走缓存，同一个标的在短时间内不做重复分析。这套组合下来，单标的成本从五万Token降到了一万以内，而输出质量几乎没有变化。

5. 从0到1的实践路径与个人建议

5.1 新手保持节奏的几个练手项目

如果你刚接触AI Agent，我建议不要一上来就照着淘宝数据搭建多Agent分析系统，先做三个小项目练手。

第一个：单Agent新闻问答。只做一个新闻采集模块加一个Prompt，让它回答“最近三天有哪些关于某公司的重要新闻，分别是什么方向”。这个项目能帮助你熟悉大模型调用、JSON解析和基本的缓存逻辑。

第二个：RAG检索问答。把公司公告文本切片存入向量库，让Agent根据用户问题检索相关段落再回答。这个项目能帮你建立起对上下文窗口和检索质量的直觉。

第三个：双Agent协作。一个Agent负责找数据，一个Agent负责审核数据，再加一个简单的LangGraph流程把它们连起来。到了这一步，你基本就掌握了多Agent协作的核心。

做完这三个项目再动工做完整的投研系统，你的难度感受会完全不一样。

5.2 低代码平台与自建方案怎么选

现在市面上也有不少Agent搭建平台，比如扣子、Dify这类可视化编排工具，适合做原型验证和中小规模场景。我用过一段时间，最大的感受是上手快、内置了常用的模型网关和工具节点，非常适合验证你的提示词和业务流程设计是否合理。

但如果你的需求像投资研究这样，涉及大量自定义数据源、自建爬虫、高频定时任务和复杂的并发控制，我还是推荐自建。平台通常会限制你的数据接入方式和执行环境，而投资研究恰恰在数据源和合规控制上要求很高。我的建议是：先用平台把流程跑通，确认业务逻辑后再切到自建代码方案。

5.3 关于期货交易方向的一些提醒

有几个朋友问过我：既然能做股票研究，是不是也能做个AI Agent直接做期货交易？技术上确实有空间，期货的行情数据结构更规范，很多接口都是现成的。但我个人的真实建议是：交易执行和研究辅助完全是两码事，不要混在一起。

研究辅助的目标是输出“可能性”和“风险点”，错了顶多浪费一点精力；但自动交易执行出错了，是真金白银的亏损。期货还有保证金、强平、滑点、手续费这些问题，任何一个没处理好，趋势判断再准也可能亏在交易环节。我自己目前把AI Agent定位在“研究副驾”而不是“自动司机”——所有交易决策仍然需要人确认。

另外，个人搭建全自动程序化交易系统还涉及合规报备和账户门槛的问题，各个平台的要求不一样。建议想往这个方向走的朋友，先把行情接入、信号生成、研究报告输出这套研究链路做扎实，再考虑要不要碰执行端。

我个人做这套系统的最大收获，并不是省了多少小时盯盘，而是把过去依赖经验和情绪的判断，变成了一套可以审计、可以回放、可以不断改进的流程。每一个结论都有出处，每一次分析都留痕，这种“可回溯性”本身就非常有价值。后面我还在计划把历史报告归档、把分析结果跟实际行情走势做批量对照，慢慢建立起自己的策略复盘数据库。如果你也在做类似的项目，很欢迎一起交流。