作與LangGraph實戰(zhàn)解析)
做投資研究這幾年我最大的感受就是信息量太大一個人根本看不過來。所以我從去年開始嘗試把AI Agent引入投資研究工作流用多智能體分工處理數(shù)據(jù)采集、新聞輿情、財報分析和風險總結把原來每天要花兩三個小時的盯盤和翻資料壓縮成十分鐘就能拿到一份帶完整邏輯鏈和來源引用的結構化報告。這篇文章就把我的經(jīng)驗整體整理一遍從架構設計、技術選型到關鍵代碼、踩坑記錄都會講到最后還有一套給新手的從0到1練手路徑。適合正在做量化投研、準備用Agent替代重復勞動的朋友參考。1. 為什么選擇用AI Agent重構投資研究流程1.1 傳統(tǒng)投資研究方式的真實痛點先說清楚我要解決的痛點。做股票、期貨這類二級市場的投研本質上是在跟信息賽跑而信息是無限膨脹的。以前我的流程大致是早晨起來先看隔夜外盤再刷一遍財經(jīng)新聞然后翻公告、研報最后自己手工匯總成當天的觀察清單。這套流程有兩個致命的毛病第一是覆蓋不全。一個人精力的上限擺在那里盯了三五個行業(yè)其他行業(yè)冒出的重要變化根本沒精力處理。我試過同時跟蹤二十只股票每只股票每天光公告、新聞、大宗交易記錄就能產(chǎn)生幾十條信息這還沒算技術面的價格形態(tài)。堅持了兩周就發(fā)現(xiàn)所謂跟蹤基本變成了“看標題式關心”很多關鍵信息其實都漏過去了。第二是情緒干擾。人天生容易受賬戶浮盈浮虧影響漲了容易樂觀跌了容易悲觀。這導致同一個消息在不同情緒狀態(tài)下你解讀出來的“風險等級”完全不一樣。更麻煩的是這種偏差你自己很難察覺。后來我意識到如果能把信息收集、初步分析、風險提示這些重復勞動交給程序自己只做決策層的工作就能大幅減少這種情緒污染。1.2 我期待Agent方案解決什么問題開始動手前我給自己列了幾個明確的目標信息采集要全分析過程要快結論要能回溯。所謂“回溯”就是說任何一個結論比如“這只股票最近負面輿情增多”你點開之后要能看到它依據(jù)了哪幾條新聞、哪幾份公告而不是黑盒模型告訴你一個結論就沒有然后了。這也是我最終放棄“直接調用大模型API問幾個問題”這種簡單方案的原因。單次問答模式看起來能用但做不到持續(xù)跟蹤和多人協(xié)作更做不到按標準流程批量處理幾十只股票。AI Agent恰恰適合這種場景——它可以把“搜集數(shù)據(jù)—分析基本面—評估新聞情緒—生成報告”拆成多個步驟每一步都有明確輸入輸出還能動態(tài)調整。相當于我請了四個研究員各自守著一塊任務最后把分析結果匯總到我這里。1.3 Agent方案和RPA、普通腳本的本質區(qū)別有人可能會說這不就是爬蟲加腳本嗎我之前也做過純爬蟲方案寫死了采集頻率和規(guī)則結果每次業(yè)務邏輯一調整就要改一堆代碼。Agent方案最大的不同在于它有推理和規(guī)劃能力。還是用新聞分析舉例子。RPA腳本能做的極限是抓取新聞列表按關鍵詞過濾然后輸出一個“提到XX公司多少條”的統(tǒng)計。但AI Agent能做的是先識別出新聞里涉及的公司、事件類型、影響方向再根據(jù)信源權威性加權最后結合當前股價位置給出“這個消息已經(jīng)被price in多少”的推測。這些步驟RPA沒法用固定規(guī)則寫死因為每一條新聞的表達方式都不同只有具備語義理解能力的模型才能在一個動態(tài)流程里完成。2. 整體架構設計與核心流程2.1 系統(tǒng)整體架構分層整個項目我分成了四層數(shù)據(jù)接入層、分析Agent層、流程編排層、服務輸出層。每一層的職責都很明確沒有把邏輯混在一起。層級職責核心組件數(shù)據(jù)接入層行情、新聞、公告、財務數(shù)據(jù)的采集與清洗AkShare、Tushare、自建爬蟲、httpx異步客戶端分析Agent層分工完成基本面、技術面、輿情、風險分析LangChain工具調用、各Agent提示詞模板流程編排層管理多Agent執(zhí)行順序、條件分支和狀態(tài)傳遞LangGraph StateGraph服務輸出層對外提供HTTP接口、任務調度與結果緩存FastAPI、Redis、Celery這套分層的出發(fā)點是可替換性。比如今天AkShare接口改版我只需要改數(shù)據(jù)接入層里的一個小模塊分析Agent和編排層完全不用動。同樣的道理今天想從GPT換到DeepSeek或Qwen只需要在模型工廠里改一個配置項所有Agent的代碼框架保持不變。2.2 數(shù)據(jù)源選型與接入策略數(shù)據(jù)源是整個項目的地基這里踩的坑最多。我的選型邏輯很簡單能用免費穩(wěn)定接口解決的不用爬蟲能一次拿全量數(shù)據(jù)的不要逐條請求。行情數(shù)據(jù)我主要用AkShare和Tushare兩個庫。AkShare勝在接口全、更新快從A股實時行情到期貨夜盤數(shù)據(jù)都有覆蓋Tushare勝在數(shù)據(jù)結構規(guī)范、權限體系完整做歷史回測時更可靠。我的實測經(jīng)驗是日常實時行情用AkShare就夠但如果要拉五年以上的日線數(shù)據(jù)做回測Tushare的積分接口更穩(wěn)。新聞類數(shù)據(jù)是最麻煩的。財經(jīng)網(wǎng)站有反爬機制而且不同網(wǎng)站的欄目結構經(jīng)常變。我的做法是自建一個小型采集服務定時抓取幾個可信度比較高的財經(jīng)門戶的財經(jīng)頭條和個股新聞板塊存到數(shù)據(jù)庫后做去重和清洗。這里特別提醒一句抓取公開網(wǎng)頁要注意目標網(wǎng)站的robots協(xié)議和訪問頻率別把別人的站點打掛了。我這邊控制單IP請求頻率在每秒兩次以內同時設置請求失敗自動退避。公告數(shù)據(jù)我用巨潮資訊網(wǎng)的公開接口這個源的好處是權威、格式統(tǒng)一。財務數(shù)據(jù)則在財報季從AkShare對準報告期批量拉取。宏觀數(shù)據(jù)如果不涉及敏感細分項也可以用公開的統(tǒng)計接口但我不做預測只做參考這點后面會細說。2.3 分析Agent如何分工我設計了五個Agent每只只負責一件事這樣提示詞可以寫得非常聚焦不會出現(xiàn)一個Agent什么都會但什么都做不精的情況。新聞輿情Agent負責抓取目標標的相關新聞做情感極性判斷和熱度趨勢分析。基本面Agent負責讀財報指標比如營收增速、凈利率、資產(chǎn)負債率判斷與上一報告期的變化情況。技術面Agent負責計算均線、MACD、量價配合度生成短期趨勢描述。估值Agent負責把當前市盈率、市凈率放到歷史區(qū)間里輸出“處于歷史什么水位”的判斷。綜合風控Agent負責匯總前面四個Agent的輸出檢查是否有矛盾點生成最終的風險提示和總結報告。這種分工還有一個好處就是單Agent的上下文窗口占用非常低不至于把新聞全文一股腦塞給一個模型。金融大模型本來就容易在長上下文里丟失關鍵信息切短輸入以后準確率提升很明顯。3. 關鍵實現(xiàn)與實操記錄3.1 環(huán)境準備與項目結構整個項目基于FastAPI和LangGraph構建模型層用LangChain的OpenAI兼容接口方便隨時切換不同廠商的模型。如果你要復現(xiàn)環(huán)境依賴大概是這樣pip install fastapi uvicorn langchain langchain-openai langgraph pip install akshare tushare pandas redis asyncpg這些依賴不代表你全部都要用AkShare的數(shù)據(jù)如果只走同步接口不裝asyncpg也沒關系。我的原則是先讓流程跑通再考慮性能和并發(fā)不要在第一步就引入過多依賴。項目目錄我按模塊組織入口很清晰investment_research/ ├── main.py # FastAPI入口 ├── agents/ # 各Agent的提示詞和調用邏輯 │ ├── news_agent.py │ ├── fundamental_agent.py │ ├── technical_agent.py │ └── risk_agent.py ├── datasources/ # 數(shù)據(jù)接入層 │ ├── market.py │ └── news.py ├── workflows/ # LangGraph編排 │ └── research_graph.py └── services/ # 業(yè)務服務層 └── research_service.py別看目錄小我后來加了定時任務模塊、緩存模塊、回調告警模塊目錄結構基本還能維持住靠的就是一開始把“數(shù)據(jù)”和“分析邏輯”徹底分離。3.2 數(shù)據(jù)接入模塊實現(xiàn)先拿行情數(shù)據(jù)舉例。AkShare的實時行情接口返回的是全市場快照數(shù)據(jù)量很大我每次只提取目標代碼那一行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: fsymbol {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[成交額]), }這個接口實測第一次調用會比較慢因為要拉全市場數(shù)據(jù)所以我加了緩存——同一分鐘內請求同一只股票直接返回上一次結果避免反復全表掃描。新聞采集也類似抓下來的標題和正文會存入本地SQLite再做一次簡單的去重按股票代碼建立索引。3.3 新聞輿情Agent的構建輿情Agent是整個系統(tǒng)里最有用的一個也是提示詞寫得最細致的。我給它設計了三個輸出維度情感得分、影響領域、熱度趨勢。from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate model ChatOpenAI( modelqwen-plus, temperature0.2, max_tokens512, ) prompt ChatPromptTemplate.from_messages([ (system, 你是專業(yè)的金融輿情分析師。請根據(jù)下面的新聞標題和摘要分析這條新聞對指定上市公司的影響。 要求 1. 先判斷情感傾向輸出positive/neutral/negative之一。 2. 給出一個-1到1之間的情感分數(shù)正數(shù)代表利好。 3. 判斷影響領域只能從【業(yè)績】【管理層】【行業(yè)政策】【市場情緒】【供應鏈】中選一個最匹配的。 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)核心點在于“必須引用新聞原文中的關鍵句”。加了這個約束之后模型編造理由的情況少了很多就算它判斷錯了我也能順著它引用的句子去復核不至于出現(xiàn)完全找不到依據(jù)的結論。3.4 用LangGraph編排多Agent流程LangGraph對我來說最大的價值是狀態(tài)傳遞和條件分支。它不像LangChain原來的Chain那樣固定順序而是能根據(jù)前的節(jié)點輸出決定下一步走向這個特性在投研場景里太實用了。舉個實際例子基本面Agent如果發(fā)現(xiàn)最新財報季數(shù)據(jù)還沒披露就跳到“等待季報模板”分支不再去做無意義的盈利預測如果新聞輿情Agent發(fā)現(xiàn)負面新聞數(shù)量超過閾值風控Agent就會被提前觸發(fā)在最終報告里加大風險警告權重。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之后就可以直接傳初始狀態(tài)運行了。LangGraph還會自動把每一步的狀態(tài)變化記錄下來這對事后復盤特別重要——我可以清楚看到一條結論是在哪一步、基于什么數(shù)據(jù)產(chǎn)生的。3.5 FastAPI接口與并發(fā)扛量實踐熱搜里有個詞很準“AI Agent怎么扛并發(fā)”。我實際開發(fā)中確實被并發(fā)問題教育過。一開始我只是簡單把流程串起來單線程跑然后發(fā)現(xiàn)兩個問題一是同步請求大模型太慢一個標的完整流程要一二十秒二是多個請求同時進來程序直接卡死。我的解決方案分幾層。首先是FastAPI接口把同步函數(shù)改成異步并且用后臺任務模式處理“接收請求—立即返回任務ID—后臺跑研究流程—完成后主動推送給用戶”。這樣用戶體驗好很多不用干等。import asyncio from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel app FastAPI() sem asyncio.Semaphore(10) # 控制同時執(zhí)行的Agent任務數(shù) 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配額。不同廠商的模型都有速率限制一旦超限就會被限流甚至封禁。我的做法是維護一個信號量限制同時進行的大模型調用數(shù)另外加了一層帶指數(shù)退避的重試機制。如果有連續(xù)失敗就把任務丟進延遲隊列避免雪崩。除了接口層Redis緩存也很關鍵。我做了兩級緩存第一級是對數(shù)據(jù)源的緩存AkShare全市場行情兩分鐘內只拉一次第二級是對分析結果的緩存同一只股票在股票池里短時間重復請求直接用上次生成的報告不再重復調用大模型。實測下來并發(fā)從5路撐到30路沒有太大問題成本也省了一半多。4. 踩坑與排查實錄4.1 大模型幻覺金融場景尤其嚴重不管用什么模型投資研究里最大的風險就是幻覺。金融數(shù)據(jù)講究精確到小數(shù)點模型如果編一個假的毛利率出來那整個分析就廢了。我踩過最離譜的一次是讓基本面Agent總結某公司的盈利情況模型信誓旦旦地寫“凈利潤同比大幅增長35%”我拿原始財報一核對實際數(shù)據(jù)是下降12%。后來我加了兩個防線一是所有數(shù)字必須帶出處模型必須給出它引用的公告編號或字段名稱二是增加一個獨立的校驗Agent專門去對比模型輸出和結構化財務數(shù)據(jù)不一致就標記為“數(shù)據(jù)沖突”并重新生成。加了這兩道防線之后基本再沒出現(xiàn)過明顯的數(shù)字幻覺。4.2 上下文塞爆與Token超限新聞數(shù)量一多把所有正文塞進Prompt肯定不行。有一次我跟蹤一只熱門股票當天相關新聞六十多條全塞進去直接超過了模型的上下文窗口跑出一個報錯。這個問題我用MapReduce的方式解決。先讓模型對每條新聞只抽“情感標簽影響領域關鍵句”不寫長分析然后匯總所有標簽再做整體情緒判斷。這樣每條新聞最多消耗兩三百個Token六十條也才一萬多Token完全在可控范圍。再往后新聞量更大時我用向量數(shù)據(jù)庫做過一版先把新聞嵌入存儲分析時只取跟當前標的相關度最高的二十條效果更穩(wěn)。4.3 Agent跑飛、死循環(huán)和超時多Agent編排最擔心的就是流程跑飛。LangGraph雖然提供了狀態(tài)機能力但節(jié)點之間如果某個工具返回了意外格式節(jié)點函數(shù)沒有做好容錯整個流程就會卡在中間節(jié)點不回傳。我遇到過一個典型問題新聞采集服務被目標網(wǎng)站反爬攔截返回了空列表。新聞Agent拿到空列表后繼續(xù)往下一個節(jié)點傳風控Agent發(fā)現(xiàn)“沒有新聞”竟然給出“輿情平穩(wěn)”的結論——這等于把數(shù)據(jù)缺失誤判成利好很危險。修復方案是每個Agent節(jié)點加校驗輸入數(shù)據(jù)不滿足最小數(shù)量時必須在結論里標注“數(shù)據(jù)缺失可信度降級”并且不允許直接把缺失當正常。另外一定要設置LangGraph的recursion_limit和每個節(jié)點的執(zhí)行超時。我最初沒設超時有一次模型服務超限導致節(jié)點重試了二十多次白白浪費了很長時間。后來統(tǒng)一設定為單節(jié)點最多重試3次、單次等待不超過30秒跑飛的情況基本絕跡。4.4 成本控制單次分析燒掉的Token比想象中多投資研究的Agent流程會反復調用大模型如果設計不當成本會高得嚇人。我第一版跑完整流程時單只股票的分析居然要消耗五萬Token一個月跟蹤二十只股票費用直接超預算。優(yōu)化思路主要有三條。第一條能用小模型解決的絕不用大模型新聞情感分類這類簡單任務用7B量級的模型就很好綜合報告這種需要深度推理的才上最強的模型。第二條多Agent之間傳遞中間結果時只傳結構化摘要不傳原始文本減少Token量。第三條相似的查詢請求走緩存同一個標的在短時間內不做重復分析。這套組合下來單標的成本從五萬Token降到了一萬以內而輸出質量幾乎沒有變化。5. 從0到1的實踐路徑與個人建議5.1 新手保持節(jié)奏的幾個練手項目如果你剛接觸AI Agent我建議不要一上來就照著淘寶數(shù)據(jù)搭建多Agent分析系統(tǒng)先做三個小項目練手。第一個單Agent新聞問答。只做一個新聞采集模塊加一個Prompt讓它回答“最近三天有哪些關于某公司的重要新聞分別是什么方向”。這個項目能幫助你熟悉大模型調用、JSON解析和基本的緩存邏輯。第二個RAG檢索問答。把公司公告文本切片存入向量庫讓Agent根據(jù)用戶問題檢索相關段落再回答。這個項目能幫你建立起對上下文窗口和檢索質量的直覺。第三個雙Agent協(xié)作。一個Agent負責找數(shù)據(jù)一個Agent負責審核數(shù)據(jù)再加一個簡單的LangGraph流程把它們連起來。到了這一步你基本就掌握了多Agent協(xié)作的核心。做完這三個項目再動工做完整的投研系統(tǒng)你的難度感受會完全不一樣。5.2 低代碼平臺與自建方案怎么選現(xiàn)在市面上也有不少Agent搭建平臺比如扣子、Dify這類可視化編排工具適合做原型驗證和中小規(guī)模場景。我用過一段時間最大的感受是上手快、內置了常用的模型網(wǎng)關和工具節(jié)點非常適合驗證你的提示詞和業(yè)務流程設計是否合理。但如果你的需求像投資研究這樣涉及大量自定義數(shù)據(jù)源、自建爬蟲、高頻定時任務和復雜的并發(fā)控制我還是推薦自建。平臺通常會限制你的數(shù)據(jù)接入方式和執(zhí)行環(huán)境而投資研究恰恰在數(shù)據(jù)源和合規(guī)控制上要求很高。我的建議是先用平臺把流程跑通確認業(yè)務邏輯后再切到自建代碼方案。5.3 關于期貨交易方向的一些提醒有幾個朋友問過我既然能做股票研究是不是也能做個AI Agent直接做期貨交易技術上確實有空間期貨的行情數(shù)據(jù)結構更規(guī)范很多接口都是現(xiàn)成的。但我個人的真實建議是交易執(zhí)行和研究輔助完全是兩碼事不要混在一起。研究輔助的目標是輸出“可能性”和“風險點”錯了頂多浪費一點精力但自動交易執(zhí)行出錯了是真金白銀的虧損。期貨還有保證金、強平、滑點、手續(xù)費這些問題任何一個沒處理好趨勢判斷再準也可能虧在交易環(huán)節(jié)。我自己目前把AI Agent定位在“研究副駕”而不是“自動司機”——所有交易決策仍然需要人確認。另外個人搭建全自動程序化交易系統(tǒng)還涉及合規(guī)報備和賬戶門檻的問題各個平臺的要求不一樣。建議想往這個方向走的朋友先把行情接入、信號生成、研究報告輸出這套研究鏈路做扎實再考慮要不要碰執(zhí)行端。我個人做這套系統(tǒng)的最大收獲并不是省了多少小時盯盤而是把過去依賴經(jīng)驗和情緒的判斷變成了一套可以審計、可以回放、可以不斷改進的流程。每一個結論都有出處每一次分析都留痕這種“可回溯性”本身就非常有價值。后面我還在計劃把歷史報告歸檔、把分析結果跟實際行情走勢做批量對照慢慢建立起自己的策略復盤數(shù)據(jù)庫。如果你也在做類似的項目很歡迎一起交流。 AI Agent重構投資研究：多智能體協(xié)作與LangGraph實戰(zhàn)解析
做投資研究這幾年我最大的感受就是信息量太大一個人根本看不過來。所以我從去年開始嘗試把AI Agent引入投資研究工作流用多智能體分工處理數(shù)據(jù)采集、新聞輿情、財報分析和風險總結把原來每天要花兩三個小時的盯盤和翻資料壓縮成十分鐘就能拿到一份帶完整邏輯鏈和來源引用的結構化報告。這篇文章就把我的經(jīng)驗整體整理一遍從架構設計、技術選型到關鍵代碼、踩坑記錄都會講到最后還有一套給新手的從0到1練手路徑。適合正在做量化投研、準備用Agent替代重復勞動的朋友參考。1. 為什么選擇用AI Agent重構投資研究流程1.1 傳統(tǒng)投資研究方式的真實痛點先說清楚我要解決的痛點。做股票、期貨這類二級市場的投研本質上是在跟信息賽跑而信息是無限膨脹的。以前我的流程大致是早晨起來先看隔夜外盤再刷一遍財經(jīng)新聞然后翻公告、研報最后自己手工匯總成當天的觀察清單。這套流程有兩個致命的毛病第一是覆蓋不全。一個人精力的上限擺在那里盯了三五個行業(yè)其他行業(yè)冒出的重要變化根本沒精力處理。我試過同時跟蹤二十只股票每只股票每天光公告、新聞、大宗交易記錄就能產(chǎn)生幾十條信息這還沒算技術面的價格形態(tài)。堅持了兩周就發(fā)現(xiàn)所謂跟蹤基本變成了“看標題式關心”很多關鍵信息其實都漏過去了。第二是情緒干擾。人天生容易受賬戶浮盈浮虧影響漲了容易樂觀跌了容易悲觀。這導致同一個消息在不同情緒狀態(tài)下你解讀出來的“風險等級”完全不一樣。更麻煩的是這種偏差你自己很難察覺。后來我意識到如果能把信息收集、初步分析、風險提示這些重復勞動交給程序自己只做決策層的工作就能大幅減少這種情緒污染。1.2 我期待Agent方案解決什么問題開始動手前我給自己列了幾個明確的目標信息采集要全分析過程要快結論要能回溯。所謂“回溯”就是說任何一個結論比如“這只股票最近負面輿情增多”你點開之后要能看到它依據(jù)了哪幾條新聞、哪幾份公告而不是黑盒模型告訴你一個結論就沒有然后了。這也是我最終放棄“直接調用大模型API問幾個問題”這種簡單方案的原因。單次問答模式看起來能用但做不到持續(xù)跟蹤和多人協(xié)作更做不到按標準流程批量處理幾十只股票。AI Agent恰恰適合這種場景——它可以把“搜集數(shù)據(jù)—分析基本面—評估新聞情緒—生成報告”拆成多個步驟每一步都有明確輸入輸出還能動態(tài)調整。相當于我請了四個研究員各自守著一塊任務最后把分析結果匯總到我這里。1.3 Agent方案和RPA、普通腳本的本質區(qū)別有人可能會說這不就是爬蟲加腳本嗎我之前也做過純爬蟲方案寫死了采集頻率和規(guī)則結果每次業(yè)務邏輯一調整就要改一堆代碼。Agent方案最大的不同在于它有推理和規(guī)劃能力。還是用新聞分析舉例子。RPA腳本能做的極限是抓取新聞列表按關鍵詞過濾然后輸出一個“提到XX公司多少條”的統(tǒng)計。但AI Agent能做的是先識別出新聞里涉及的公司、事件類型、影響方向再根據(jù)信源權威性加權最后結合當前股價位置給出“這個消息已經(jīng)被price in多少”的推測。這些步驟RPA沒法用固定規(guī)則寫死因為每一條新聞的表達方式都不同只有具備語義理解能力的模型才能在一個動態(tài)流程里完成。2. 整體架構設計與核心流程2.1 系統(tǒng)整體架構分層整個項目我分成了四層數(shù)據(jù)接入層、分析Agent層、流程編排層、服務輸出層。每一層的職責都很明確沒有把邏輯混在一起。層級職責核心組件數(shù)據(jù)接入層行情、新聞、公告、財務數(shù)據(jù)的采集與清洗AkShare、Tushare、自建爬蟲、httpx異步客戶端分析Agent層分工完成基本面、技術面、輿情、風險分析LangChain工具調用、各Agent提示詞模板流程編排層管理多Agent執(zhí)行順序、條件分支和狀態(tài)傳遞LangGraph StateGraph服務輸出層對外提供HTTP接口、任務調度與結果緩存FastAPI、Redis、Celery這套分層的出發(fā)點是可替換性。比如今天AkShare接口改版我只需要改數(shù)據(jù)接入層里的一個小模塊分析Agent和編排層完全不用動。同樣的道理今天想從GPT換到DeepSeek或Qwen只需要在模型工廠里改一個配置項所有Agent的代碼框架保持不變。2.2 數(shù)據(jù)源選型與接入策略數(shù)據(jù)源是整個項目的地基這里踩的坑最多。我的選型邏輯很簡單能用免費穩(wěn)定接口解決的不用爬蟲能一次拿全量數(shù)據(jù)的不要逐條請求。行情數(shù)據(jù)我主要用AkShare和Tushare兩個庫。AkShare勝在接口全、更新快從A股實時行情到期貨夜盤數(shù)據(jù)都有覆蓋Tushare勝在數(shù)據(jù)結構規(guī)范、權限體系完整做歷史回測時更可靠。我的實測經(jīng)驗是日常實時行情用AkShare就夠但如果要拉五年以上的日線數(shù)據(jù)做回測Tushare的積分接口更穩(wěn)。新聞類數(shù)據(jù)是最麻煩的。財經(jīng)網(wǎng)站有反爬機制而且不同網(wǎng)站的欄目結構經(jīng)常變。我的做法是自建一個小型采集服務定時抓取幾個可信度比較高的財經(jīng)門戶的財經(jīng)頭條和個股新聞板塊存到數(shù)據(jù)庫后做去重和清洗。這里特別提醒一句抓取公開網(wǎng)頁要注意目標網(wǎng)站的robots協(xié)議和訪問頻率別把別人的站點打掛了。我這邊控制單IP請求頻率在每秒兩次以內同時設置請求失敗自動退避。公告數(shù)據(jù)我用巨潮資訊網(wǎng)的公開接口這個源的好處是權威、格式統(tǒng)一。財務數(shù)據(jù)則在財報季從AkShare對準報告期批量拉取。宏觀數(shù)據(jù)如果不涉及敏感細分項也可以用公開的統(tǒng)計接口但我不做預測只做參考這點后面會細說。2.3 分析Agent如何分工我設計了五個Agent每只只負責一件事這樣提示詞可以寫得非常聚焦不會出現(xiàn)一個Agent什么都會但什么都做不精的情況。新聞輿情Agent負責抓取目標標的相關新聞做情感極性判斷和熱度趨勢分析。基本面Agent負責讀財報指標比如營收增速、凈利率、資產(chǎn)負債率判斷與上一報告期的變化情況。技術面Agent負責計算均線、MACD、量價配合度生成短期趨勢描述。估值Agent負責把當前市盈率、市凈率放到歷史區(qū)間里輸出“處于歷史什么水位”的判斷。綜合風控Agent負責匯總前面四個Agent的輸出檢查是否有矛盾點生成最終的風險提示和總結報告。這種分工還有一個好處就是單Agent的上下文窗口占用非常低不至于把新聞全文一股腦塞給一個模型。金融大模型本來就容易在長上下文里丟失關鍵信息切短輸入以后準確率提升很明顯。3. 關鍵實現(xiàn)與實操記錄3.1 環(huán)境準備與項目結構整個項目基于FastAPI和LangGraph構建模型層用LangChain的OpenAI兼容接口方便隨時切換不同廠商的模型。如果你要復現(xiàn)環(huán)境依賴大概是這樣pip install fastapi uvicorn langchain langchain-openai langgraph pip install akshare tushare pandas redis asyncpg這些依賴不代表你全部都要用AkShare的數(shù)據(jù)如果只走同步接口不裝asyncpg也沒關系。我的原則是先讓流程跑通再考慮性能和并發(fā)不要在第一步就引入過多依賴。項目目錄我按模塊組織入口很清晰investment_research/ ├── main.py # FastAPI入口 ├── agents/ # 各Agent的提示詞和調用邏輯 │ ├── news_agent.py │ ├── fundamental_agent.py │ ├── technical_agent.py │ └── risk_agent.py ├── datasources/ # 數(shù)據(jù)接入層 │ ├── market.py │ └── news.py ├── workflows/ # LangGraph編排 │ └── research_graph.py └── services/ # 業(yè)務服務層 └── research_service.py別看目錄小我后來加了定時任務模塊、緩存模塊、回調告警模塊目錄結構基本還能維持住靠的就是一開始把“數(shù)據(jù)”和“分析邏輯”徹底分離。3.2 數(shù)據(jù)接入模塊實現(xiàn)先拿行情數(shù)據(jù)舉例。AkShare的實時行情接口返回的是全市場快照數(shù)據(jù)量很大我每次只提取目標代碼那一行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: fsymbol {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[成交額]), }這個接口實測第一次調用會比較慢因為要拉全市場數(shù)據(jù)所以我加了緩存——同一分鐘內請求同一只股票直接返回上一次結果避免反復全表掃描。新聞采集也類似抓下來的標題和正文會存入本地SQLite再做一次簡單的去重按股票代碼建立索引。3.3 新聞輿情Agent的構建輿情Agent是整個系統(tǒng)里最有用的一個也是提示詞寫得最細致的。我給它設計了三個輸出維度情感得分、影響領域、熱度趨勢。from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate model ChatOpenAI( modelqwen-plus, temperature0.2, max_tokens512, ) prompt ChatPromptTemplate.from_messages([ (system, 你是專業(yè)的金融輿情分析師。請根據(jù)下面的新聞標題和摘要分析這條新聞對指定上市公司的影響。 要求 1. 先判斷情感傾向輸出positive/neutral/negative之一。 2. 給出一個-1到1之間的情感分數(shù)正數(shù)代表利好。 3. 判斷影響領域只能從【業(yè)績】【管理層】【行業(yè)政策】【市場情緒】【供應鏈】中選一個最匹配的。 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)核心點在于“必須引用新聞原文中的關鍵句”。加了這個約束之后模型編造理由的情況少了很多就算它判斷錯了我也能順著它引用的句子去復核不至于出現(xiàn)完全找不到依據(jù)的結論。3.4 用LangGraph編排多Agent流程LangGraph對我來說最大的價值是狀態(tài)傳遞和條件分支。它不像LangChain原來的Chain那樣固定順序而是能根據(jù)前的節(jié)點輸出決定下一步走向這個特性在投研場景里太實用了。舉個實際例子基本面Agent如果發(fā)現(xiàn)最新財報季數(shù)據(jù)還沒披露就跳到“等待季報模板”分支不再去做無意義的盈利預測如果新聞輿情Agent發(fā)現(xiàn)負面新聞數(shù)量超過閾值風控Agent就會被提前觸發(fā)在最終報告里加大風險警告權重。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之后就可以直接傳初始狀態(tài)運行了。LangGraph還會自動把每一步的狀態(tài)變化記錄下來這對事后復盤特別重要——我可以清楚看到一條結論是在哪一步、基于什么數(shù)據(jù)產(chǎn)生的。3.5 FastAPI接口與并發(fā)扛量實踐熱搜里有個詞很準“AI Agent怎么扛并發(fā)”。我實際開發(fā)中確實被并發(fā)問題教育過。一開始我只是簡單把流程串起來單線程跑然后發(fā)現(xiàn)兩個問題一是同步請求大模型太慢一個標的完整流程要一二十秒二是多個請求同時進來程序直接卡死。我的解決方案分幾層。首先是FastAPI接口把同步函數(shù)改成異步并且用后臺任務模式處理“接收請求—立即返回任務ID—后臺跑研究流程—完成后主動推送給用戶”。這樣用戶體驗好很多不用干等。import asyncio from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel app FastAPI() sem asyncio.Semaphore(10) # 控制同時執(zhí)行的Agent任務數(shù) 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配額。不同廠商的模型都有速率限制一旦超限就會被限流甚至封禁。我的做法是維護一個信號量限制同時進行的大模型調用數(shù)另外加了一層帶指數(shù)退避的重試機制。如果有連續(xù)失敗就把任務丟進延遲隊列避免雪崩。除了接口層Redis緩存也很關鍵。我做了兩級緩存第一級是對數(shù)據(jù)源的緩存AkShare全市場行情兩分鐘內只拉一次第二級是對分析結果的緩存同一只股票在股票池里短時間重復請求直接用上次生成的報告不再重復調用大模型。實測下來并發(fā)從5路撐到30路沒有太大問題成本也省了一半多。4. 踩坑與排查實錄4.1 大模型幻覺金融場景尤其嚴重不管用什么模型投資研究里最大的風險就是幻覺。金融數(shù)據(jù)講究精確到小數(shù)點模型如果編一個假的毛利率出來那整個分析就廢了。我踩過最離譜的一次是讓基本面Agent總結某公司的盈利情況模型信誓旦旦地寫“凈利潤同比大幅增長35%”我拿原始財報一核對實際數(shù)據(jù)是下降12%。后來我加了兩個防線一是所有數(shù)字必須帶出處模型必須給出它引用的公告編號或字段名稱二是增加一個獨立的校驗Agent專門去對比模型輸出和結構化財務數(shù)據(jù)不一致就標記為“數(shù)據(jù)沖突”并重新生成。加了這兩道防線之后基本再沒出現(xiàn)過明顯的數(shù)字幻覺。4.2 上下文塞爆與Token超限新聞數(shù)量一多把所有正文塞進Prompt肯定不行。有一次我跟蹤一只熱門股票當天相關新聞六十多條全塞進去直接超過了模型的上下文窗口跑出一個報錯。這個問題我用MapReduce的方式解決。先讓模型對每條新聞只抽“情感標簽影響領域關鍵句”不寫長分析然后匯總所有標簽再做整體情緒判斷。這樣每條新聞最多消耗兩三百個Token六十條也才一萬多Token完全在可控范圍。再往后新聞量更大時我用向量數(shù)據(jù)庫做過一版先把新聞嵌入存儲分析時只取跟當前標的相關度最高的二十條效果更穩(wěn)。4.3 Agent跑飛、死循環(huán)和超時多Agent編排最擔心的就是流程跑飛。LangGraph雖然提供了狀態(tài)機能力但節(jié)點之間如果某個工具返回了意外格式節(jié)點函數(shù)沒有做好容錯整個流程就會卡在中間節(jié)點不回傳。我遇到過一個典型問題新聞采集服務被目標網(wǎng)站反爬攔截返回了空列表。新聞Agent拿到空列表后繼續(xù)往下一個節(jié)點傳風控Agent發(fā)現(xiàn)“沒有新聞”竟然給出“輿情平穩(wěn)”的結論——這等于把數(shù)據(jù)缺失誤判成利好很危險。修復方案是每個Agent節(jié)點加校驗輸入數(shù)據(jù)不滿足最小數(shù)量時必須在結論里標注“數(shù)據(jù)缺失可信度降級”并且不允許直接把缺失當正常。另外一定要設置LangGraph的recursion_limit和每個節(jié)點的執(zhí)行超時。我最初沒設超時有一次模型服務超限導致節(jié)點重試了二十多次白白浪費了很長時間。后來統(tǒng)一設定為單節(jié)點最多重試3次、單次等待不超過30秒跑飛的情況基本絕跡。4.4 成本控制單次分析燒掉的Token比想象中多投資研究的Agent流程會反復調用大模型如果設計不當成本會高得嚇人。我第一版跑完整流程時單只股票的分析居然要消耗五萬Token一個月跟蹤二十只股票費用直接超預算。優(yōu)化思路主要有三條。第一條能用小模型解決的絕不用大模型新聞情感分類這類簡單任務用7B量級的模型就很好綜合報告這種需要深度推理的才上最強的模型。第二條多Agent之間傳遞中間結果時只傳結構化摘要不傳原始文本減少Token量。第三條相似的查詢請求走緩存同一個標的在短時間內不做重復分析。這套組合下來單標的成本從五萬Token降到了一萬以內而輸出質量幾乎沒有變化。5. 從0到1的實踐路徑與個人建議5.1 新手保持節(jié)奏的幾個練手項目如果你剛接觸AI Agent我建議不要一上來就照著淘寶數(shù)據(jù)搭建多Agent分析系統(tǒng)先做三個小項目練手。第一個單Agent新聞問答。只做一個新聞采集模塊加一個Prompt讓它回答“最近三天有哪些關于某公司的重要新聞分別是什么方向”。這個項目能幫助你熟悉大模型調用、JSON解析和基本的緩存邏輯。第二個RAG檢索問答。把公司公告文本切片存入向量庫讓Agent根據(jù)用戶問題檢索相關段落再回答。這個項目能幫你建立起對上下文窗口和檢索質量的直覺。第三個雙Agent協(xié)作。一個Agent負責找數(shù)據(jù)一個Agent負責審核數(shù)據(jù)再加一個簡單的LangGraph流程把它們連起來。到了這一步你基本就掌握了多Agent協(xié)作的核心。做完這三個項目再動工做完整的投研系統(tǒng)你的難度感受會完全不一樣。5.2 低代碼平臺與自建方案怎么選現(xiàn)在市面上也有不少Agent搭建平臺比如扣子、Dify這類可視化編排工具適合做原型驗證和中小規(guī)模場景。我用過一段時間最大的感受是上手快、內置了常用的模型網(wǎng)關和工具節(jié)點非常適合驗證你的提示詞和業(yè)務流程設計是否合理。但如果你的需求像投資研究這樣涉及大量自定義數(shù)據(jù)源、自建爬蟲、高頻定時任務和復雜的并發(fā)控制我還是推薦自建。平臺通常會限制你的數(shù)據(jù)接入方式和執(zhí)行環(huán)境而投資研究恰恰在數(shù)據(jù)源和合規(guī)控制上要求很高。我的建議是先用平臺把流程跑通確認業(yè)務邏輯后再切到自建代碼方案。5.3 關于期貨交易方向的一些提醒有幾個朋友問過我既然能做股票研究是不是也能做個AI Agent直接做期貨交易技術上確實有空間期貨的行情數(shù)據(jù)結構更規(guī)范很多接口都是現(xiàn)成的。但我個人的真實建議是交易執(zhí)行和研究輔助完全是兩碼事不要混在一起。研究輔助的目標是輸出“可能性”和“風險點”錯了頂多浪費一點精力但自動交易執(zhí)行出錯了是真金白銀的虧損。期貨還有保證金、強平、滑點、手續(xù)費這些問題任何一個沒處理好趨勢判斷再準也可能虧在交易環(huán)節(jié)。我自己目前把AI Agent定位在“研究副駕”而不是“自動司機”——所有交易決策仍然需要人確認。另外個人搭建全自動程序化交易系統(tǒng)還涉及合規(guī)報備和賬戶門檻的問題各個平臺的要求不一樣。建議想往這個方向走的朋友先把行情接入、信號生成、研究報告輸出這套研究鏈路做扎實再考慮要不要碰執(zhí)行端。我個人做這套系統(tǒng)的最大收獲并不是省了多少小時盯盤而是把過去依賴經(jīng)驗和情緒的判斷變成了一套可以審計、可以回放、可以不斷改進的流程。每一個結論都有出處每一次分析都留痕這種“可回溯性”本身就非常有價值。后面我還在計劃把歷史報告歸檔、把分析結果跟實際行情走勢做批量對照慢慢建立起自己的策略復盤數(shù)據(jù)庫。如果你也在做類似的項目很歡迎一起交流。