使用 ReAct 和 LangGraph 构建股票研究智能体
LangGraph ReAct智能体会获取一只股票的代码，按既定顺序调用基本面分析、新闻资讯、技术分析以及分析师工具，同时在人类干预的间隔中撰写简短分析报告。
来源: Building a Financial Research Agent with ReAct, LangGraph, and LangChain
在市场开盘前研究个股是一项常见的任务。那些关注印度股市中如ICICI、HDFC、Reliance或HAL等股票的人，往往需要花费40分钟在Screener.in、Moneycontrol以及各类新闻页面之间切换，手动整理信息。这类重复性的工作恰恰是可以由智能代理来承担的。
以下是基于ReAct模式构建的完整可运行研究代理实现，全部为实际代码而非伪代码。只需输入如RELIANCE.NS或AAPL这样的股票代码，该代理便会综合分析基本面、新闻情绪、技术指标以及分析师观点，进而生成结构化的投资报告。
令人惊讶的是，并不存在固定的脚本——代理会根据所获取的信息自行决定研究的内容和顺序，这种自适应的推理方式正是ReAct模式的精髓所在。
智能代理的功能
给定一个股票代码后，该智能代理会独立完成以下工作：
- 获取基本面数据：市盈率、营收增长率、每股收益、负债率以及利润率。
- 收集最新新闻，并结合理由将其评定为看涨、看跌或中性。
- 计算技术指标：RSI、MACD、50日与200日移动平均线以及成交量趋势。
- 获取分析师共识，即买入、持有和卖出评级的分布情况。
- 将所有信息整合成一份结构化的分析报告，最后给出投资建议。
工具调用的顺序由智能体自行决定。它可能先分析技术面数据，再考虑基本面因素；或者先阅读新闻，发现重大企业事件后立刻查看财务数据，以确认相关数据是否支持该事件。这样的决策使得ReAct智能体有别于传统链式智能体。整体设计如下：
User: "Research RELIANCE.NS"
│
▼
┌─────────────┐ tool_call ┌─────────────┐
│ Agent Node │ ──────────────────► │ Tools Node │
│ (LLM) │ ◄────────────────── │ (ToolNode) │
└─────────────┘ observation └─────────────┘
│
│ no more tool calls
▼
Final Investment Brief
这就是推理、行动、观察，然后再推理的ReAct循环，该过程会不断重复，直到智能体认为已获得足够信息[3]。
准备步骤
首先安装依赖项：
pip install langgraph langchain langchain-openai yfinance pandas-ta \
tavily-python python-dotenv pydantic
接下来是包含 API 密钥的 .env 文件：
OPENAI_API_KEY=your_openai_key
TAVILY_API_KEY=your_tavily_key
该项目的布局较小：
financial_agent/
├── .env
├── state.py # Agent state definition
├── tools.py # All 4 research tools
├── agent.py # LLM + tool binding
├── graph.py # LangGraph graph definition
└── main.py # Entry point
智能体状态
状态是在图中的每个节点间传递的共享内存。在LangGraph [1]中，每个节点都会读取该状态并向其写入数据，因此它便充当了整个研究过程中智能体的工作内存。
# state.py
from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
# The full message history - drives the ReAct loop
# add_messages is a reducer: appends new messages rather than overwriting
messages: Annotated[list[BaseMessage], add_messages]
# The ticker being researched
ticker: str
# Structured research outputs (populated by tools)
financials: dict
news_sentiment: dict
technical_indicators: dict
analyst_ratings: dict
# Final output
investment_brief: str
有两个要点值得注意。messages字段使用add_messages作为归约函数[1]，因此每次更新都会在原有消息后追加新内容而非替换它们。这样就能在整个循环过程中保留完整的“思考、行动、观察”历史记录。如果没有这个归约函数，每个节点都会覆盖之前的对话内容，导致智能体忘记自己之前已经检查过什么，这种错误通常表现为反复调用相同的工具。而financials和news_sentiment这类结构化研究字段则只是普通的字典，会在工具运行时被填充数据，并在后续推理中作为参考使用。
工具
工具定义可能比图表结构或系统提示更为重要。大语言模型会阅读每个工具的描述，以决定何时调用哪个工具。模糊的描述会导致某些工具被跳过或调用顺序异常，因此这类描述最好像为经验丰富的同事撰写的文档，而非供验证工具使用的架构说明。
# tools.py
import json
from datetime import datetime, timedelta
import pandas_ta as ta
import yfinance as yf
from langchain_core.tools import tool
from tavily import TavilyClient
tavily = TavilyClient()
@tool
def get_stock_financials(ticker: str) -> str:
"""
Fetch key fundamental financial metrics for a stock ticker.
Returns P/E ratio, forward P/E, revenue growth (YoY), EPS (TTM),
profit margin, debt-to-equity ratio, and free cash flow yield.
Use this tool first to understand whether a company's fundamentals
are strong before diving into price-based analysis.
Works for both NSE tickers (e.g. RELIANCE.NS) and US tickers (e.g. AAPL).
"""
try:
stock = yf.Ticker(ticker) # [4]
info = stock.info
fundamentals = {
"company_name": info.get("longName", ticker),
"sector": info.get("sector", "N/A"),
"market_cap_b": round(info.get("marketCap", 0) / 1e9, 2),
"pe_ratio_ttm": info.get("trailingPE"),
"pe_ratio_forward": info.get("forwardPE"),
"eps_ttm": info.get("trailingEps"),
"revenue_growth_yoy": info.get("revenueGrowth"),
"earnings_growth_yoy": info.get("earningsGrowth"),
"profit_margin": info.get("profitMargins"),
"operating_margin": info.get("operatingMargins"),
"debt_to_equity": info.get("debtToEquity"),
"current_ratio": info.get("currentRatio"),
"return_on_equity": info.get("returnOnEquity"),
"free_cashflow_b": round(info.get("freeCashflow", 0) / 1e9, 2),
"dividend_yield": info.get("dividendYield"),
"52_week_high": info.get("fiftyTwoWeekHigh"),
"52_week_low": info.get("fiftyTwoWeekLow"),
"current_price": info.get("currentPrice"),
}
# Flag key concerns automatically
concerns = []
if fundamentals["debt_to_equity"] and fundamentals["debt_to_equity"] > 200:
concerns.append("High debt-to-equity ratio - leverage risk")
if fundamentals["profit_margin"] and fundamentals["profit_margin"] < 0:
concerns.append("Negative profit margin - company is unprofitable")
if fundamentals["revenue_growth_yoy"] and fundamentals["revenue_growth_yoy"] < -0.05:
concerns.append("Revenue declining year-over-year")
fundamentals["flagged_concerns"] = concerns
return json.dumps(fundamentals, indent=2)
except Exception as e:
return json.dumps({"error": str(e), "ticker": ticker})
@tool
def get_news_sentiment(ticker: str) -> str:
"""
Search for recent news about a stock ticker and analyse overall sentiment.
Returns the top 5 recent headlines with publication dates and URLs,
a sentiment classification (BULLISH / BEARISH / NEUTRAL), and
a brief reasoning for the sentiment call.
Use this tool to understand current market narrative and any recent
events (earnings, regulatory issues, product launches, macro exposure)
that could affect the stock's near-term trajectory.
"""
try:
stock = yf.Ticker(ticker)
company_name = stock.info.get("longName", ticker)
# Search for recent news via Tavily [6]
results = tavily.search(
query=f"{company_name} {ticker} stock news analysis 2025 2026",
max_results=5,
search_depth="advanced",
)
headlines = []
content_for_sentiment = []
for r in results.get("results", []):
headlines.append({
"title": r.get("title", ""),
"url": r.get("url", ""),
"published_date": r.get("published_date", ""),
"snippet": r.get("content", "")[:300],
})
content_for_sentiment.append(r.get("content", ""))
# Simple keyword-based sentiment scoring
combined_text = " ".join(content_for_sentiment).lower()
bullish_keywords = [
"beat", "exceeded", "strong", "growth", "upgrade", "buy",
"outperform", "record", "surge", "rally", "positive", "robust"
]
bearish_keywords = [
"miss", "disappointed", "weak", "decline", "downgrade", "sell",
"underperform", "concern", "risk", "drop", "fell", "negative",
"lawsuit", "investigation", "layoff"
]
bullish_score = sum(combined_text.count(kw) for kw in bullish_keywords)
bearish_score = sum(combined_text.count(kw) for kw in bearish_keywords)
if bullish_score > bearish_score * 1.5:
sentiment = "BULLISH"
sentiment_reasoning = f"News skews positive ({bullish_score} bullish signals vs {bearish_score} bearish)"
elif bearish_score > bullish_score * 1.5:
sentiment = "BEARISH"
sentiment_reasoning = f"News skews negative ({bearish_score} bearish signals vs {bullish_score} bullish)"
else:
sentiment = "NEUTRAL"
sentiment_reasoning = f"Mixed signals ({bullish_score} bullish, {bearish_score} bearish)"
output = {
"ticker": ticker,
"company": company_name,
"headlines": headlines,
"overall_sentiment": sentiment,
"sentiment_reasoning": sentiment_reasoning,
"bullish_signal_count": bullish_score,
"bearish_signal_count": bearish_score,
}
return json.dumps(output, indent=2)
except Exception as e:
return json.dumps({"error": str(e), "ticker": ticker})
@tool
def get_technical_indicators(ticker: str) -> str:
"""
Calculate technical analysis indicators for a stock using the last
200 days of price data. Returns RSI (14-day), MACD and signal line,
50-day and 200-day SMAs, Golden/Death Cross status, and volume analysis.
Use this tool to assess whether the stock is in a technically
favourable or unfavourable position for entry.
RSI above 70 = overbought. RSI below 30 = oversold.
Golden cross (SMA50 > SMA200) is bullish; death cross is bearish.
"""
try:
stock = yf.Ticker(ticker)
df = stock.history(period="200d") # [4]
if df.empty or len(df) < 50:
return json.dumps({"error": "Insufficient price history", "ticker": ticker})
# RSI [5]
df.ta.rsi(length=14, append=True)
rsi = round(df["RSI_14"].iloc[-1], 2)
# MACD [5]
df.ta.macd(fast=12, slow=26, signal=9, append=True)
macd_val = round(df["MACD_12_26_9"].iloc[-1], 4)
macd_signal = round(df["MACDs_12_26_9"].iloc[-1], 4)
macd_histogram = round(df["MACDh_12_26_9"].iloc[-1], 4)
# SMAs [5]
df.ta.sma(length=50, append=True)
df.ta.sma(length=200, append=True)
sma_50 = round(df["SMA_50"].iloc[-1], 2)
sma_200 = round(df["SMA_200"].iloc[-1], 2) if len(df) >= 200 else None
current_price = round(df["Close"].iloc[-1], 2)
# Golden/Death Cross
if sma_200:
cross_status = (
"GOLDEN CROSS (bullish - SMA50 above SMA200)"
if sma_50 > sma_200
else "DEATH CROSS (bearish - SMA50 below SMA200)"
)
else:
cross_status = "N/A (insufficient data for SMA200)"
# Volume
avg_volume_30d = int(df["Volume"].tail(30).mean())
current_volume = int(df["Volume"].iloc[-1])
volume_ratio = round(current_volume / avg_volume_30d, 2)
rsi_signal = (
"OVERBOUGHT - potential reversal ahead" if rsi > 70
else "OVERSOLD - potential buying opportunity" if rsi < 30
else f"NEUTRAL ({rsi})"
)
macd_signal_str = (
"BULLISH - MACD above signal line"
if macd_val > macd_signal
else "BEARISH - MACD below signal line"
)
technicals = {
"ticker": ticker,
"current_price": current_price,
"rsi_14": rsi,
"rsi_signal": rsi_signal,
"macd": macd_val,
"macd_signal_line": macd_signal,
"macd_histogram": macd_histogram,
"macd_interpretation": macd_signal_str,
"sma_50": sma_50,
"sma_200": sma_200,
"cross_status": cross_status,
"price_vs_sma50_pct": round((current_price - sma_50) / sma_50 * 100, 2),
"volume_30d_avg": avg_volume_30d,
"volume_today": current_volume,
"volume_ratio_vs_avg": volume_ratio,
"volume_note": (
"Above average" if volume_ratio > 1.2
else "Below average" if volume_ratio < 0.8
else "In line with average"
),
}
return json.dumps(technicals, indent=2)
except Exception as e:
return json.dumps({"error": str(e), "ticker": ticker})
@tool
def get_analyst_ratings(ticker: str) -> str:
"""
Fetch current Wall Street / institutional analyst consensus ratings for a stock.
Returns the distribution of Strong Buy / Buy / Hold / Sell / Strong Sell
recommendations, the consensus, and the mean price target with implied
upside or downside from the current price.
Use this tool to understand what professional analysts collectively
think about the stock's near-term potential.
"""
try:
stock = yf.Ticker(ticker) # [4]
info = stock.info
current_price = info.get("currentPrice", 0)
target_mean = info.get("targetMeanPrice")
target_high = info.get("targetHighPrice")
target_low = info.get("targetLowPrice")
recommendation = info.get("recommendationKey", "N/A").upper()
num_analysts = info.get("numberOfAnalystOpinions", 0)
implied_upside = None
if target_mean and current_price:
implied_upside = round((target_mean - current_price) / current_price * 100, 2)
rec_df = stock.recommendations
rec_summary = {}
if rec_df is not None and not rec_df.empty:
latest = rec_df.tail(1)
for col in ["strongBuy", "buy", "hold", "sell", "strongSell"]:
if col in latest.columns:
rec_summary[col] = int(latest[col].values[0])
output = {
"ticker": ticker,
"current_price": current_price,
"consensus_recommendation": recommendation,
"number_of_analysts": num_analysts,
"price_target_mean": target_mean,
"price_target_high": target_high,
"price_target_low": target_low,
"implied_upside_pct": implied_upside,
"ratings_distribution": rec_summary,
}
return json.dumps(output, indent=2)
except Exception as e:
return json.dumps({"error": str(e), "ticker": ticker})
# Register all tools for the agent
TOOLS = [
get_stock_financials,
get_news_sentiment,
get_technical_indicators,
get_analyst_ratings,
]
此处的情绪评分采用较为基础的方法，仅通过统计关键词出现频率来实现。在实际应用中，会用基于大语言模型的情绪分析模块或经过微调的分类器来替代。为展示ReAct循环的功能，这个简化版本已足够满足需求，同时还能降低依赖性。
系统提示与智能体
系统提示决定了智能体的推理方式：包括其处理优先级、顺序以及最终输出的格式。精确的提示至关重要，因为模糊的提示会导致模糊的推理结果。
# agent.py
from langchain_core.messages import SystemMessage
from langchain_openai import ChatOpenAI
from tools import TOOLS
SYSTEM_PROMPT = """You are an expert financial research analyst with deep experience
in fundamental analysis, technical analysis, and market sentiment.
Your task is to produce a comprehensive investment brief for a given stock ticker.
RESEARCH APPROACH:
1. Start with fundamentals - understand the business quality first
2. Then check news sentiment - what is the current market narrative?
3. Run technical analysis - is the price action favourable for entry?
4. Finally check analyst consensus - how does Wall Street see it?
You have access to four research tools. Use ALL of them before writing your brief.
Do not skip any tool - each provides a different dimension of analysis.
INVESTMENT BRIEF FORMAT (use this exact structure in your final response):
## Investment Brief: [TICKER] - [COMPANY NAME]
**Date:** [today's date]
**Recommendation:** [STRONG BUY / BUY / HOLD / SELL / STRONG SELL]
**Conviction:** [HIGH / MEDIUM / LOW]
### The Bull Case
[2-3 sentences on why someone would buy this stock]
### The Bear Case
[2-3 sentences on the key risks and why someone would avoid it]
### Fundamental Snapshot
[Key metrics: P/E, growth, margins, debt. 3-4 bullet points]
### Technical Picture
[RSI, MACD, moving averages, volume. 2-3 bullet points]
### News Sentiment
[Current sentiment and what is driving it. 2-3 bullet points]
### Analyst Consensus
[What analysts think. Price target. 1-2 bullet points]
### Final Verdict
[3-4 sentences synthesising everything into a clear recommendation with reasoning]
---
Be direct. Be specific. Use actual numbers from your research.
Avoid generic statements - every claim must be backed by data you retrieved.
"""
def create_agent():
llm = ChatOpenAI(
model="gpt-4o",
temperature=0.1, # Low temperature for consistent analytical output
)
# Bind tools - this injects tool schemas into every LLM call
return llm.bind_tools(TOOLS)
图表
在这一点上，LangGraph [1][2] 相较于标准的 LangChain AgentExecutor 具有明显优势。该图表能够实现对推理过程的精确控制：可以查看每一次状态转换，可添加中断以便人工审查，还能在不影响现有逻辑的情况下新增节点。
# graph.py
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import END, START, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition
from agent import SYSTEM_PROMPT, create_agent
from state import AgentState
from tools import TOOLS
def agent_node(state: AgentState) -> dict:
"""
Core reasoning node. Calls the LLM with the full message history.
The LLM either:
- Returns a tool call → routes to tools node
- Returns a final text response → routes to END
"""
llm_with_tools = create_agent()
messages = state["messages"]
if not any(isinstance(m, SystemMessage) for m in messages):
messages = [SystemMessage(content=SYSTEM_PROMPT)] + messages
response = llm_with_tools.invoke(messages)
return {"messages": [response]}
def build_graph() -> StateGraph:
"""
Construct and compile the ReAct agent graph.
Flow:
START → agent_node → (tool call?) → tools_node → agent_node → ... → END
"""
graph_builder = StateGraph(AgentState)
graph_builder.add_node("agent", agent_node)
graph_builder.add_node("tools", ToolNode(TOOLS))
graph_builder.add_edge(START, "agent")
# tools_condition checks if the last message has a tool_call.
# If yes → "tools". If no → END.
graph_builder.add_conditional_edges("agent", tools_condition)
# After tools execute, always return to the agent for the next reasoning step
graph_builder.add_edge("tools", "agent")
return graph_builder.compile()
tools_condition函数[1]默默地完成了许多工作。在每个智能体步骤之后，它都会查看最新的消息。来自大语言模型的tool_call会被路由到tools节点，而纯文本——即模型已完成推理并准备输出的内容——则会被路由到END。
并没有预设的步骤次数。循环会持续进行，直到智能体满足条件，对于新闻更新频繁的市场行情，可能需要五到六轮，而对于较为平静的市场，则大约需要三轮。入口点将这一切串联起来：
# main.py
import sys
from dotenv import load_dotenv
from langchain_core.messages import HumanMessage
from graph import build_graph
load_dotenv()
def research_stock(ticker: str, stream: bool = True) -> str:
"""
Run the financial research agent on a given ticker.
Args:
ticker: Stock symbol - e.g. "AAPL", "RELIANCE.NS", "VEDL.NS"
stream: If True, prints the ReAct loop steps in real-time
Returns:
Final investment brief as a string
"""
graph = build_graph()
initial_state = {
"messages": [
HumanMessage(
content=f"Please research {ticker.upper()} and produce a comprehensive "
f"investment brief using all available research tools."
)
],
"ticker": ticker.upper(),
"financials": {},
"news_sentiment": {},
"technical_indicators": {},
"analyst_ratings": {},
"investment_brief": "",
}
final_brief = ""
if stream:
print(f"\n{'='*60}")
print(f" Financial Research Agent - {ticker.upper()}")
print(f"{'='*60}\n")
for event in graph.stream(initial_state, stream_mode="values"):
messages = event.get("messages", [])
if messages:
last_msg = messages[-1]
msg_type = type(last_msg).__name__
if msg_type == "AIMessage":
if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
for tc in last_msg.tool_calls:
print(f"🔧 Calling tool: {tc['name']}")
print(f" Args: {tc['args']}\n")
elif last_msg.content:
print("📋 INVESTMENT BRIEF\n")
print(last_msg.content)
final_brief = last_msg.content
elif msg_type == "ToolMessage":
print(f"✅ Tool result received: {last_msg.name}")
content_preview = last_msg.content[:200].replace('\n', ' ')
print(f" Preview: {content_preview}...\n")
else:
result = graph.invoke(initial_state)
final_brief = result["messages"][-1].content
return final_brief
if __name__ == "__main__":
ticker = sys.argv[1] if len(sys.argv) > 1 else "RELIANCE.NS"
research_stock(ticker, stream=True)
运行它
该代理可通过命令行针对任意市场进行运行：
# Indian equity
python main.py RELIANCE.NS
# US equity
python main.py AAPL
流式输出如下所示：
============================================================
Financial Research Agent — RELIANCE.NS
============================================================
🔧 Calling tool: get_stock_financials
Args: {'ticker': 'RELIANCE.NS'}
✅ Tool result received: get_stock_financials
Preview: { "company_name": "Reliance Industries Limited",
"sector": "Energy", "market_cap_b": 198.4, "pe_ratio_ttm": 24.1 ...
🔧 Calling tool: get_news_sentiment
Args: {'ticker': 'RELIANCE.NS'}
✅ Tool result received: get_news_sentiment
Preview: { "overall_sentiment": "BULLISH", "sentiment_reasoning":
"News skews positive (42 bullish signals vs 11 bearish) ...
🔧 Calling tool: get_technical_indicators
Args: {'ticker': 'RELIANCE.NS'}
✅ Tool result received: get_technical_indicators
Preview: { "rsi_14": 55.7, "rsi_signal": "NEUTRAL (55.7)",
"cross_status": "GOLDEN CROSS (bullish - SMA50 above SMA200)" ...
🔧 Calling tool: get_analyst_ratings
Args: {'ticker': 'RELIANCE.NS'}
✅ Tool result received: get_analyst_ratings
Preview: { "consensus_recommendation": "BUY", "number_of_analysts": 32,
"implied_upside_pct": 18.6 ...
📋 INVESTMENT BRIEF
## Investment Brief: RELIANCE.NS - Reliance Industries Limited
**Date:** March 22, 2026
**Recommendation:** BUY
**Conviction:** MEDIUM
...
在撰写简报之前，该智能体调用了全部四种工具，并且不会仅在完成基础步骤后就停止。这正是ReAct正常运作的方式：推理、行动、观察，再重复推理，直至满足要求[3]。
人类介入机制
LangGraph [1][2]的一个突出功能是能够在运行过程中暂停图处理，等待人类介入。在金融领域这一功能非常实用，因为在简报最终确定之前可能需要进行合规性检查，或者有人希望在研究进行到一半时给予指导。
# In graph.py
from langgraph.checkpoint.memory import MemorySaver
def build_graph_with_hitl() -> StateGraph:
graph_builder = StateGraph(AgentState)
graph_builder.add_node("agent", agent_node)
graph_builder.add_node("tools", ToolNode(TOOLS))
graph_builder.add_edge(START, "agent")
graph_builder.add_conditional_edges("agent", tools_condition)
graph_builder.add_edge("tools", "agent")
checkpointer = MemorySaver()
return graph_builder.compile(
checkpointer=checkpointer,
interrupt_before=["agent"], # Pause before the final synthesis step
)
借助持久化状态，操作员可以查看已收集的内容，添加指令后再继续操作：
config = {"configurable": {"thread_id": "reliance-research-001"}}
graph = build_graph_with_hitl()
# Run until the interrupt
events = list(graph.stream(initial_state, config=config))
# Inspect intermediate state
state = graph.get_state(config)
print("Last message:", state.values["messages"][-1].content)
# Redirect the agent's focus before it writes the brief
graph.update_state(
config,
{"messages": [HumanMessage(
"Please pay particular attention to Reliance's Jio and retail "
"segment revenue contributions in your final brief."
)]}
)
# Resume to completion
for event in graph.stream(None, config=config):
print(event)
标准的 AgentExecutor 无法干净地实现这一点。基于检查点的持久化机制使得在执行过程中进行干预成为一项核心功能，而非临时解决方案。
扩展图结构
图结构使得新增元素变得简单。有两个方向值得考虑。
第一种是风险评估节点。它位于各工具之后、最终汇报之前，用于交叉验证结果并标记出复合风险：
def risk_assessment_node(state: AgentState) -> dict:
"""Compute a risk score from gathered research before writing the brief."""
risk_flags = []
if state.get("financials"):
fins = state["financials"]
if fins.get("debt_to_equity", 0) > 200:
risk_flags.append("HIGH_LEVERAGE")
if fins.get("revenue_growth_yoy", 0) < -0.05:
risk_flags.append("DECLINING_REVENUE")
# Cross-reference: overbought on bad news is a high-risk combination
if (state.get("news_sentiment", {}).get("overall_sentiment") == "BEARISH"
and state.get("technical_indicators", {}).get("rsi_14", 50) > 65):
risk_flags.append("OVERBOUGHT_ON_BAD_NEWS")
risk_message = f"Risk flags: {risk_flags}" if risk_flags else "No major risk flags."
return {"messages": [HumanMessage(content=f"[Risk Assessment]: {risk_message}")]}
第二种是同行比较工具。对于印度市场的大盘股而言，一个显而易见的问题是：与同行业其他公司相比，它的股价是偏低还是偏高？
@tool
def get_peer_comparison(ticker: str) -> str:
"""
Compare the stock's valuation multiples (P/E, P/S, EV/EBITDA) against
its top 5 sector peers. Returns whether the stock trades at a premium
or discount to peers on key metrics. Use this when you need to contextualise
whether the current valuation is justified relative to competitors.
"""
# Implementation using yfinance sector data
...
只需在TOOLS中注册即可，代理会在下次运行时自动识别它，而无需更改图表。
代理与链
“ReAct代理”这一术语的使用较为随意，因此有必要明确其区别。链是一种固定顺序：第一步之后是第二步，再之后是第三步，具体步骤由第一步的结果决定。其结构在设计阶段就已确定。
ReAct代理则会在每次获取观测结果后决定下一步该做什么[3]。假设get_stock_financials返回的是一家收入持续下降且无盈利能力的公司，一个优秀的代理会据此调整后续操作，比如调用get_news_sentiment来查看市场是否有好转的迹象。如果get_technical_indicators显示一家基本面良好的公司其RSI指标严重超卖，它可能会直接调用get_analyst_ratings来确认专业人士是否也注意到了这一状况。
这种自适应排序源自“思考、行动、观察”循环，因此在流式输出中能够体现出来——因为智能体会在每次工具调用之间进行推理，而不仅仅在最后一步。
它还有实际优势：意外输入不会导致系统崩溃。没有分析师覆盖的小盘股也不会使其出错，因为智能体会注意到这一差异并据此进行推理。在财报季期间伴有大量新闻的个股，其置信度评分会表现出更高的不确定性。正是这种灵活性，使得人们愿意接受更复杂的结构，而非简单的链式结构。
每个文件的功能
state.py中包含带有add_messages约简器的AgentStateTypedDict [1]。如果去掉该约简器，智能体将在每一步丢失上下文，不得不重新处理工作。
tools.py 定义了基于 yfinance [4] 和 pandas-ta [5] 的四种研究工具。大语言模型在选择工具时会读取这些函数的文档字符串，因此文档的质量会直接影响选择的准确性。agent.py 将这些工具与 GPT-4o 相结合，并使用结构化的系统提示语来规范输出格式。graph.py 利用 tools_condition 构建 StateGraph [1][2]，以实现动态的 ReAct 路由。该过程没有固定的迭代次数，会在智能体表示已完成时终止。main.py 以流式方式执行任务，从而实时展示整个“思考、行动、观察”的循环过程。“人在回路”扩展功能展示了为何在需要处理真实投资组合时，选择 LangGraph [2] 比自行实现 ReAct 循环更值得。持久化状态、检查点中断以及运行中的干预功能，都是在实际应用中至关重要的特性。
参考文献
[1] LangGraph 关于 StateGraph、ToolNode、tools_condition 以及 add_messages 的文档，地址为 docs.langchain.com/langgraph。
[2] LangGraph 代码仓库，地址为 github.com/langchain-ai/langgraph。
[3] Yao 等人于 2023 年发表的关于在语言模型中结合推理与行动能力的论文，地址为 arxiv.org/abs/2210.03629。
[4] 用于下载雅虎财经数据的库 yfinance，地址为 pypi.org/project/yfinance。
[5] 用于 pandas 的技术分析库 pandas-ta，地址为 pypi.org/project/pandas-ta。
[6] 专为大型语言模型应用优化的搜索 API Tavily，地址为 tavily.com。