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