AI Agent 自动化投资
🤖 AI Agent = LLM + 工具 + 记忆 + 规划 —— 从"问答助手"升级为"自动执行的投资决策者"。
🎯 AI Agent 核心架构
📌 Agent = LLM + 4 大组件
┌─────────────────────────────────────────┐
│ AI Agent 架构 │
├─────────────────────────────────────────┤
│ LLM (大脑) │
│ ├── GPT-4 / Claude / Gemini / Qwen │
│ ├── 推理、规划、决策 │
│ └── 多模态（文本/图像/音频/视频） │
│ │
│ Tools (工具) │
│ ├── API: Tushare / AKShare / 雪球 │
│ ├── 数据库: SQL / Vector │
│ ├── 浏览器: 抓取新闻 / 公告 │
│ ├── 代码执行: Python / SQL │
│ └── 交易接口: 券商 API │
│ │
│ Memory (记忆) │
│ ├── 短期: 对话上下文 │
│ ├── 长期: 向量数据库 / 用户偏好 │
│ └── 情景: 经验案例库 │
│ │
│ Planning (规划) │
│ ├── 任务分解: ReAct / CoT │
│ ├── 反思: Reflection / Self-Critique │
│ └── 多步推理: Tree of Thought │
└─────────────────────────────────────────┘📌 Agent vs 普通 LLM
| 维度 | 普通 LLM | AI Agent |
|---|---|---|
| 输入 | 单次提示 | 长期记忆 + 上下文 |
| 输出 | 文本 | 行动（API 调用 / 代码执行） |
| 能力 | 知识 | 执行 |
| 工具 | 无 | 多工具组合 |
| 规划 | 无 | 自主规划 |
| 错误恢复 | 无 | 反思 + 调整 |
| 适用场景 | 问答、写作 | 自动化、研究、交易 |
📊 Agent 关键模式
📌 ReAct（Reasoning + Acting）
问题：上海茅台 vs 腾讯控股 哪个更值得买？
Thought 1: 用户想比较两家公司，需要 PE / 增速 / 现金流 数据
Action 1: 搜索"贵州茅台 PE 2024"
Observation 1: 贵州茅台 2024 PE 25.6
Thought 2: 也需要腾讯的 PE
Action 2: 搜索"腾讯控股 PE 2024"
Observation 2: 腾讯 2024 PE 18.5
Thought 3: 两者增速差异
Action 3: 搜索"贵州茅台 营收增速 2024"
Observation 3: 茅台 2024 营收增速 15%
Thought 4: 腾讯 营收增速
Action 4: 搜索"腾讯 营收增速 2024"
Observation 4: 腾讯 2024 营收增速 8%
Thought 5: 综合比较
Action 5: 输出分析报告📌 CoT（Chain of Thought）
- 让 LLM 显式输出推理过程
- 提高数学 / 逻辑能力 30-50%
- 配合 Few-shot 效果更佳
📌 Plan-and-Execute
用户：构建一个量化策略
↓
Planner: 把任务分解为 5 步
1. 数据获取
2. 因子计算
3. 回测
4. 风险评估
5. 报告输出
↓
Executor: 逐步执行
↓
Replanner: 根据结果调整📌 Reflection（自我反思）
Agent 第一次回答：买入茅台
↓
Critic Agent 评估：理由不够充分，需要更多数据
↓
Agent 重新回答：基于 PE / ROE / 增速 详细分析📊 主流 Agent 框架
📌 LangChain
- 官网：langchain.com
- GitHub：github.com/langchain-ai/langchain
- 特点：最流行、文档全、社区大
- 核心组件：Models / Prompts / Chains / Agents / Memory / Indexes
- 优势：多模型支持、丰富工具、生态完整
- 劣势：版本迭代快、抽象层多
python
from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import Tool
from langchain_openai import ChatOpenAI
from langchain import hub
# 1. 定义工具
def get_stock_price(ticker: str) -> str:
"""获取股票价格"""
import tushare as ts
df = ts.get_realtime_quotes(ticker)
return f"当前价: {df['price'].iloc[0]}"
tools = [
Tool(
name="StockPrice",
func=get_stock_price,
description="获取股票实时价格"
)
]
# 2. 创建 Agent
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = hub.pull("hwchase17/react")
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
# 3. 执行
result = agent_executor.invoke({
"input": "贵州茅台现在多少钱？"
})
print(result["output"])📌 LangGraph
- 官网：langchain-ai.github.io/langgraph
- 特点：基于 LangChain 的图状态机
- 核心：节点 + 边 + 状态
- 适用：多步工作流、复杂状态
python
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
# 定义状态
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
next_step: str
# 定义节点
def research(state):
# 调研
return {"messages": ["调研结果..."]}
def analyze(state):
# 分析
return {"messages": ["分析结果..."]}
def decide(state):
# 决策
return {"messages": ["决策..."], "next_step": "execute"}
def execute(state):
# 执行
return {"messages": ["执行完成"]}
# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("research", research)
workflow.add_node("analyze", analyze)
workflow.add_node("decide", decide)
workflow.add_node("execute", execute)
workflow.add_edge("research", "analyze")
workflow.add_edge("analyze", "decide")
workflow.add_edge("decide", "execute")
workflow.add_edge("execute", END)
workflow.set_entry_point("research")
app = workflow.compile()
result = app.invoke({"messages": [], "next_step": ""})📌 AutoGen（微软）
- 官网：microsoft.github.io/autogen
- 特点：多智能体对话
- 核心：UserProxyAgent / AssistantAgent / GroupChat
- 适用：复杂协作任务
python
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
# 创建多个 Agent
researcher = AssistantAgent(
name="Researcher",
llm_config={"model": "gpt-4o"},
system_message="你是一个金融研究员，负责收集和分析数据"
)
analyst = AssistantAgent(
name="Analyst",
llm_config={"model": "gpt-4o"},
system_message="你是一个量化分析师，负责因子计算和回测"
)
trader = AssistantAgent(
name="Trader",
llm_config={"model": "gpt-4o"},
system_message="你是一个交易员，负责执行交易和风控"
)
user = UserProxyAgent(
name="User",
human_input_mode="TERMINATE"
)
# 群聊
groupchat = GroupChat(
agents=[user, researcher, analyst, trader],
messages=[],
max_round=10
)
manager = GroupChatManager(groupchat=groupchat, llm_config={"model": "gpt-4o"})
user.initiate_chat(manager, message="构建一个沪深 300 动量策略")📌 CrewAI
- 特点：多智能体协作框架
- 核心：Role + Goal + Backstory
- 适用：多角色任务
python
from crewai import Agent, Task, Crew
researcher = Agent(
role="高级研究员",
goal="发现 A 股最佳投资标的",
backstory="20 年 A 股研究经验",
tools=[search_tool, tushare_tool]
)
analyst = Agent(
role="量化分析师",
goal="回测并验证策略",
backstory="10 年量化经验",
tools=[backtest_tool]
)
task1 = Task(
description="分析贵州茅台的基本面",
agent=researcher
)
task2 = Task(
description="回测白酒板块过去 5 年表现",
agent=analyst
)
crew = Crew(agents=[researcher, analyst], tasks=[task1, task2])
result = crew.kickoff()📌 MCP（Model Context Protocol）
- 定义：Anthropic 2024-11 推出的开放协议
- 目的：标准化 LLM 与外部工具 / 数据源的连接
- 核心：MCP Server（暴露工具）+ MCP Client（消费工具）
- 优势：跨模型、跨工具、跨数据源
LLM Agent
↓
MCP Client（Claude Desktop / Cursor / 自建）
↓
MCP Server（数据源 / 工具）
├── Tushare MCP Server（股票数据）
├── Backtrader MCP Server（回测）
├── 雪球 MCP Server（舆情）
└── 邮件 MCP Server（通知）📊 投资领域 Agent 实战
📌 场景 1：自动研报
python
# 自动研报生成 Agent
class ReportAgent:
def __init__(self):
self.tools = [TushareTool(), NewsTool(), Filer()]
self.llm = ChatOpenAI(model="gpt-4o")
def generate(self, ticker: str):
# 1. 拉取数据
data = self.tools[0].get_financials(ticker)
news = self.tools[1].get_news(ticker, days=30)
# 2. 提取关键观点
key_points = self.llm.invoke(f"提取以下研报要点：{data}")
# 3. 撰写研报
report = self.llm.invoke(f"""
基于以下数据撰写专业研报：
财务数据：{data}
最新新闻：{news}
关键观点：{key_points}
""")
# 4. 输出 PDF
self.tools[2].save(report, f"{ticker}_report.pdf")
return report📌 场景 2：量化因子研究
python
# 因子研究 Agent
class FactorAgent:
def research(self, factor: str):
# 1. 调研学术文献
papers = self.search_arxiv(f"{factor} factor stock returns")
# 2. 实现因子
code = self.llm.invoke(f"用 Python 实现 {factor} 因子")
# 3. 回测
backtest_result = self.backtest(code)
# 4. 分析结果
analysis = self.llm.invoke(f"""
分析以下回测结果：
因子代码：{code}
回测结果：{backtest_result}
""")
return analysis📌 场景 3：实时舆情监控
python
# 舆情监控 Agent
class SentimentAgent:
def monitor(self, ticker: str):
while True:
# 1. 抓取新闻 / 微博 / 雪球
news = self.crawl(f"{ticker} 股票 今日")
# 2. 情绪分析
sentiment = self.analyze_sentiment(news)
# 3. 触发告警
if sentiment < -0.5:
self.alert(f"{ticker} 舆情负面：{news[:5]}")
time.sleep(300) # 5 分钟一次📌 场景 4：自动交易
python
# 交易 Agent
class TradingAgent:
def execute(self, signal: dict):
# 1. 风险检查
if not self.risk_check(signal):
return "风控不通过"
# 2. 资金分配
size = self.position_size(signal)
# 3. 下单
order = self.broker.order(
ticker=signal['ticker'],
side=signal['side'],
size=size
)
# 4. 记录
self.log(order)
return order📊 量化实战案例
📌 案例 1：AI 选股 Agent
python
# 综合 AI 选股 Agent
class AIStockPicker:
def __init__(self):
self.fundamental_agent = FundamentalAgent()
self.technical_agent = TechnicalAgent()
self.sentiment_agent = SentimentAgent()
self.risk_agent = RiskAgent()
self.llm = ChatOpenAI(model="gpt-4o")
def pick_stocks(self, universe: list, top_n: int = 10):
# 1. 各 Agent 评分
scores = {}
for ticker in universe:
f = self.fundamental_agent.score(ticker) # 基本面
t = self.technical_agent.score(ticker) # 技术面
s = self.sentiment_agent.score(ticker) # 舆情
r = self.risk_agent.score(ticker) # 风险
# 2. 综合评分（LLM 加权）
weights = self.llm.invoke(f"""
对以下 Agent 在当前市场环境下给股票评分：
- 基本面 (f={f}): 权重？
- 技术面 (t={t}): 权重？
- 舆情 (s={s}): 权重？
- 风险 (r={r}): 权重？
""")
scores[ticker] = f * 0.4 + t * 0.3 + s * 0.2 + r * 0.1
# 3. 排序选 Top N
top_stocks = sorted(scores.items(), key=lambda x: -x[1])[:top_n]
return top_stocks📌 案例 2：宏观 + 个股联动
python
# 宏观驱动 Agent
class MacroAgent:
def __init__(self):
self.macro_data = FredAPI() # 美联储
self.china_data = WindAPI() # 国内宏观
self.llm = ChatOpenAI(model="gpt-4o")
def analyze(self):
# 1. 拉取宏观数据
cpi = self.china_data.get_cpi()
pmi = self.china_data.get_pmi()
m2 = self.china_data.get_m2()
# 2. LLM 解读
interpretation = self.llm.invoke(f"""
解读当前宏观环境：
CPI: {cpi}（通胀）
PMI: {pmi}（经济）
M2: {m2}（货币）
给出最受益行业
""")
# 3. 联动个股
sectors = self.parse_sectors(interpretation)
stocks = self.map_sector_to_stocks(sectors)
return stocks📊 风险与挑战
📌 技术风险
| 风险 | 应对 |
|---|---|
| 幻觉 | RAG + 人工复核 |
| API 限流 | 限流 + 多源数据 |
| 数据错误 | 数据校验 + 多源对比 |
| 延迟 | 缓存 + 异步 |
| 成本 | 模型分层（小模型 + 大模型） |
📌 投资风险
| 风险 | 应对 |
|---|---|
| 策略失效 | 持续监控 + 再训练 |
| 过拟合 | 样本外测试 + 简化模型 |
| 黑天鹅 | 风控 + 止损 + 分散 |
| 流动性 | 限额 + 大单拆分 |
| 监管 | 合规审查 + 报备 |
📌 监管态势
- 2024-2025：证监会多次发布程序化交易管理规定
- 2025-04：证监会发布《证券市场程序化交易管理规定（试行）》
- 核心要求：
- 备案（高频交易必须）
- 收费（高频交易额外收费）
- 风控（异常交易监测）
- 公平（禁止频繁撤单等）
- 2026：交易所升级监管科技（RegTech）
📊 学习路径
📌 入门（0-3 月）
- Python 基础 + 异步编程
- LLM 基础（Prompt、Token、Embedding）
- LangChain 基础（Chain、Tool、Agent）
- 简单 RAG 应用
📌 进阶（3-6 月）
- LangGraph / AutoGen / CrewAI
- RAG 优化（混合检索 / 重排）
- 工具调用（Function Calling / MCP）
- 量化基础（因子、回测、风控）
📌 实战（6-12 月）
- 完整 Agent 系统（多源数据 + 多策略）
- 投资领域 RAG（研报库、新闻库）
- 自动交易（券商 API）
- 风控 + 监控 + 日志
📌 推荐资源
| 资源 | 类型 |
|---|---|
| LangChain 官方文档 | 入门 |
| DeepLearning.AI Short Courses | 入门（免费） |
| OpenAI Cookbook | 实战 |
| LangGraph 教程 | 进阶 |
| AutoGen GitHub | 多智能体 |
| Anthropic MCP 文档 | MCP |
| AI Stock Picking 实战 | 量化 |
| Quantitative Finance with Python | 量化基础 |
📖 相关阅读
- AI 选股 — 机器学习选股
- 深度学习选股 — CNN / Transformer
- 量化投资入门 — 量化基础
- 量化回测 — Backtrader
- 投资哲学 — 决策与概率
- 投资心理学 — Agent 心理偏差
💡 数据来源：LangChain / LangGraph / AutoGen 官方文档、Anthropic MCP 规范、OpenAI Cookbook、DeepLearning.AI 课程。数据截至 2026-08。
来源
- 证监会公开公告
- 上交所交易规则
- 深交所 / 北交所规则
- 中国证券业协会（SAC）、中国证券投资基金业协会（AMAC）公开资料
- 国家金融监督管理总局、人社部、财政部、税务总局公开资料
💡 数据截至 2026-08；具体规则以交易所/监管机构最新公告为准。