Создание агента для анализа акций с использованием 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.
Что делает агент
При наличии тикера агент самостоятельно выполняет следующие задачи:
- Сбор фундаментальных показателей: коэффициента P/E, темпов роста доходов, EPS, соотношения долга к капиталу и маржи.
- Сбор актуальных новостей с оценкой их влияния как положительного, отрицательного или нейтрального с обоснованием.
- Расчёт технических индикаторов: 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,
]
Оценка настроения здесь сознательно выполнена в простой форме и основывается на частоте появления ключевых слов. В реальных условиях использовалась бы оценка настроения на основе больших языковых моделей или тонко настроенный классификатор. Чтобы продемонстрировать цикл 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» используется довольно широко, поэтому необходимо четко определить разницу. Цепочка — это фиксированная последовательность: шаг 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 поиска, оптимизированный для работы с LLM, на tavily.com.