Створення агента для дослідження акцій за допомогою 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
Далі йде файл .env, який містить ключі API:
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,
]
Оцінка настрою тут навмисно є простою та ґрунтується на кількості вхідних ключових слів. У продакшн-середовищі замість неї використовуватиметься алгоритм оцінки настрою, заснований на LLM, або детально налаштований класифікатор. Щоб продемонструвати цикл 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 від LLM направляється до вузла 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 агент“ використовується досить широко, тому різниця між ними потребує точності. Ланцюжок — це фіксована послідовність: крок 1, потім 2, потім 3, незалежно від результату кроку 1. Його склад визначається ще на етапі проектування.
ReAct агент вирішує, що робити далі після кожного спостереження [3]. Припустимо, що функція get_stock_financials повертає інформацію про неприбуткову компанію зі зниженням доходів. Хороший агент враховує це під час наступних дій, можливо, виконуючи функцію get_news_sentiment, щоб з’ясувати, чи є у ринку сигнали про зміну ситуації. Якщо функція get_technical_indicators показує надмірно низький рівень RSI у компанії з міцними фундаментальними показниками, агент може одразу звернутися до функції get_analyst_ratings, щоб дізнатися, чи помітили професіонали таку саму проблему.
Така адаптивна послідовність виникає в межах циклу «Мислення, Дія, Спостереження», і її можна побачити у потоці результатів, оскільки агент робить висновки між кожним викликом інструменту, а не лише наприкінці.
Є й практична перевага. Неочікувані дані не заважають йому працювати. Компанія з невеликою капіталізацією без аналітичного охоплення не призведе до збоїв, адже агент помічає цю прогалину та знаходить способи її подолання. Індекс під час сезону звітності з інтенсивними новинами буде показувати більшу невизначеність у своєму рейтингу впевненості. Саме ця гнучкість є причиною того, чому краще прийняти додаткову складність графа, ніж просту ланцюгову структуру.
Що робить кожен файл
state.pyмістить TypedDictAgentStateразом із функцією-редукторомadd_messages[1]. Якщо видалити цей редуктор, агент втратить контекст на кожному кроці та доведеться повторювати роботу.
tools.py визначає чотири дослідницькі інструменти на основі yfinance [4] та pandas-ta [5]. Їхні документації — це те, що читає ШІ під час вибору інструменту, тому їхня якість безпосередньо впливає на точність вибору.agent.py під’єднує ці інструменти до GPT-4o та застосовує структурований системний запит, який визначає формат результату.graph.py створює StateGraph [1][2] за допомогою tools_condition для динамічного маршрутизування за принципом ReAct. Кількість ітерацій не фіксується, а виконання закінчується тоді, коли агент повідомляє про завершення.main.py транслює процес виконання, щоб весь цикл Thought, Act, Observe був видимий у реальному часі.Розширення з участю людини пояснює, чому варто обирати LangGraph [2] замість реалізованого вручну циклу ReAct. Збереження стану, переривання через чекпоїнти та втручання під час виконання — це функції, які мають значення при роботі з реальними портфелями.
Посилання
[1] Документація LangGraph щодо StateGraph, ToolNode, tools_condition та add_messages на docs.langchain.com/langgraph.
[2] Репозиторій LangGraph на github.com/langchain-ai/langgraph.
[3] Стаття про ReAct щодо поєднання міркувань та дій у мовних моделях (Yao et al., 2023) на arxiv.org/abs/2210.03629.
[4] yfinance — інструмент для завантаження даних з Yahoo Finance на pypi.org/project/yfinance.
[5] pandas-ta — бібліотека для технічного аналізу для pandas на pypi.org/project/pandas-ta.
[6] Tavily — API пошуку, налаштований для застосувань з великими мовними моделями, на tavily.com.