Агент для даследавання акцый на ReAct і LangGraph
ReAct-агент на LangGraph атрымлівае тыкер, сам выбірае парадак выкліку інструментаў фундаментальных даных, навін, тэхнічнага аналізу і аналітыкаў і піша кароткую справаздачу з паўзамі для чалавека.
Крыніцы: 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, ўособлівае ся простымі слоўнікамі, якія заполняюцца пад час выканання інструментаў і выкарыстоўваюцца падчас пазнейшага аналізу.
Інструменты
Апіранцы інструментаў можу мець большое значэнне, чым форма графа або запрас цяпеў. LLM чытае апісанне кожнага інструмента, каб вырашыць, які з іх вызваць і калі. Нечыстае апісанне прыводзіць да таго, што інструменты праўяюцца або ўпорадкуваюцца некальканосна, таму іх лепш пісаць як дакументацыю для кваліфікованага колегі, а не як схему для верыфікатора.
# 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()
The tools_condition function [1] тыхо віконуў багато працы. Пасля кожнага кроку агента яна перэвачаўць останне повідомлення. tool_call звяжаецца з вузькімі "інструментамі", тодзяк як звічайній тэкст, што означае, што мадэль завершыла міслення і готова до запісування, ведзе до 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]. Іхнія докстрынгі ўжоўваюцца LLM пад час выбору інструмента, таму якосьць гэтых докстрынгаў безпасебна вялічыць точнасць выбору.agent.py спаявае інструменты з GPT-4o і выкарыстоўвае структураваны запрос сістэмы, які фіксуе фармат выходных даных.graph.py стварае StateGraph [1][2] з tools_condition для дынамічнага керавання па прынцыпе ReAct. Колькісць ітэрацый не фіксуецца, а выконанне завершаецца, калі агент паведамляе, што ўсё завершана.main.py трансляюць выконанне, такім чынам увесь цыкл Thought, Act, Observe паказваецца ў рэальны час.Пашырэнне «Human-in-the-loop» паказвае, чаму 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 і іншыя, 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.