Tworzenie agenta do analizy akcji przy użyciu ReAct i LangGraph
Agent LangGraph ReAct bierze symbol giełdowy, wykorzystuje narzędzia analityczne oparte na danych fundamentalnych, wiadomościach, analizie technicznej oraz narzędzia analityków w ustalonym przez siebie porządku, przy czym wprowadza przerwy umożliwiające interwencję człowieka, aby stworzyć krótki raport.
Źródła: Building a Financial Research Agent with ReAct, LangGraph, and LangChain
Badanie akcji przed otwarciem rynku to powszechna, męcząca praca. Osoba obserwująca indyjskie spółki takie jak ICICI, HDFC, Reliance czy HAL może łatwo stracić 40 minut na przeglądaniu stron Screener.in, Moneycontrol oraz różnych zakładek z wiadomościami, starając się samodzielnie ułożyć obraz sytuacji. Właśnie takie powtarzalne zadania mogą zostać przejęte przez agenta.
Poniżej znajduje się kompletna, działająca implementacja agenta analitycznego opartego na wzorcu ReAct, zawierająca prawdziwy kod, a nie pseudokod. Wystarczy podać kod giełdowy takiego jak RELIANCE.NS lub AAPL. Agent analizuje podstawowe dane, nastroje rynkowe, wskaźniki techniczne oraz opinie analityków, a następnie tworzy ustrukturyzowany raport inwestycyjny.
Zaskakujące jest to, że nie istnieje żaden sztywny skrypt. Agent sam decyduje, co i w jakiej kolejności bada, na podstawie tego, co znajduje, a właśnie ta adaptacyjna forma rozumowania stanowi istotę wzorca ReAct.
Czym zajmuje się agent
Dla danego symbolu giełdowego agent samodzielnie wykonywa następujące zadania:
- Pobieranie danymi fundamentalnymi: stosunek ceny do zysku, wzrost przychodów, zysk na akcję, stosunek długu do kapitału własnego oraz marże.
- Zbieranie ostatnich wiadomości i ocenianie ich jako pozytywnych, negatywnych lub neutralnych, z uzasadnieniem.
- Obliczanie wskaźników technicznych: RSI, MACD, średnich kroczących 50 i 200 dni oraz trendu wolumenu.
- Pobieranie konsensusu analityków, czyli podział ocen na „kupuj”, „utrzymaj” i „sprzedaj”.
- Łączenie wszystkich informacji w uporządkowany raport, który kończy się rekomendacją.
Kolejność wywołań narzędzi jest decyzją agenta. Najpierw mogą pojawić się aspekty techniczne, a potem podstawowe. Albo może przeczytać wiadomości, zauważyć ważne wydarzenie korporacyjne i natychmiast sprawdzić dane finansowe, aby sprawdzić, czy liczby potwierdzają tę historię. Takie decyzje odróżniają agenta typu ReAct od zwykłej łańcuchowej struktury. Ogólny projekt:
User: "Research RELIANCE.NS"
│
▼
┌─────────────┐ tool_call ┌─────────────┐
│ Agent Node │ ──────────────────► │ Tools Node │
│ (LLM) │ ◄────────────────── │ (ToolNode) │
└─────────────┘ observation └─────────────┘
│
│ no more tool calls
▼
Final Investment Brief
To jest pętla ReAct składająca się z rozumowania, działania, obserwacji, a następnie ponownego rozumowania, powtarzana aż agent uzna, że ma wystarczająco dużo informacji [3].
Ustawienia
Najpierw zainstaluj zależności:
pip install langgraph langchain langchain-openai yfinance pandas-ta \
tavily-python python-dotenv pydantic
Następnie znajduje się plik .env zawierający klucze API:
OPENAI_API_KEY=your_openai_key
TAVILY_API_KEY=your_tavily_key
Rozmiar układu projektu jest mały:
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
Stan agenta
Stan to wspólna pamięć, która przepływa przez każdy węzeł grafu. W LangGraph [1] każdy węzeł odczytuje go i zapisuje do niego, dzięki czemu pełni rolę pamięci roboczej agenta przez całą sesję badawczą.
# 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
Dwa szczegóły wyróżniają się. Pole messages wykorzystuje add_messages jako reducer [1], więc przy każdej aktualizacji dodawane są nowe wiadomości zamiast ich zastępowania. Dzięki temu cała historia myśli, działań i obserwacji jest zachowywana przez cały cykl. Bez reducera każdy węzeł nadpisałby poprzednią rozmowę, a agent zapomniałby o tym, co sprawdził – błąd ten często objawia się wielokrotnym wywoływaniem tego samego narzędzia. Strukturyzowane pola badawcze, takie jak financials i news_sentiment, to zwykłe słowniki, które są wypełniane w miarę wykonywania narzędzi i wykorzystywane później w procesie rozumowania.
Narzędzia
Definicje narzędzi mogą być ważniejsze niż kształt grafu lub instrukcje systemowe. LLM czyta opis każdego narzędzia, aby zdecydować, które wywołać i kiedy. Niejasne opisy powodują pomijanie narzędzi lub dziwną kolejność ich wykorzystania, dlatego najlepiej jest je pisać jak dokumentację dla kompetentnego kolegi, a nie jak schemat dla walidatora.
# 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,
]
W tym przypadku ocena nastroju jest celowo prosta i opiera się na częstotliwości występowania kluczowych słów. W rzeczywistych zastosowaniach zastąpiono by ją procesem analizy nastroju opartym na modelach LLM lub dostrojonym klasyfikatorem. Aby pokazać mechanizm pętli ReAct, wystarczy prosta wersja, która sprawdza się i ogranicza zależności.
Wskazówka systemowa i agent
Wskazówka systemowa określa, w jaki sposób agent rozumuje: jego priorytety, kolejność ich realizacji oraz format końcowego wyniku. W tym przypadku precyzja ma kluczowe znaczenie, ponieważ niejasna wskazówka prowadzi do niejasnego rozumowania.
# 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)
Graf
LangGraph [1][2] zasługuje na pierwszeństwo przed standardowym LangChain AgentExecutor na tym etapie. Graf umożliwia bezpośrednią kontrolę nad pętlą rozumowania. Można zobaczyć każdą transformację stanu, dodać przerwy w celu przeglądania przez człowieka oraz dołączyć nowe węzły bez ingerencji w istniejącą logikę.
# 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()
Funkcja tools_condition [1] w tle wykonuje wiele zadań. Po każdym kroku agenta sprawdza ostatnią wiadomość. tool_call wysłany z LLM trafia do węzła tools, natomiast zwykły tekst, oznaczający, że model zakończył rozumowanie i jest gotowy do pisania, trafia do END.
Nie ma ustalonej liczby kroków. Pętla kontynuuje się, dopóki agent nie będzie zadowolony – może to być pięć lub sześć rund w przypadku giełdy z intensywnym przepływem wiadomości, a około trzech w przypadku spokojniejszej sytuacji. Punkt wejścia łączy te elementy ze sobą:
# 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)
Ruch
Agent działa z linii poleceń wobec dowolnego rynku:
# Indian equity
python main.py RELIANCE.NS
# US equity
python main.py AAPL
Wynik strumieniowania wygląda tak:
============================================================
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
...
Agent skontaktował się ze wszystkimi czterema narzędziami przed opracowaniem streszczenia i nie zatrzymał się na samych podstawach. To właśnie oznacza prawidłowe działanie ReAct: rozumowanie, działanie, obserwacja, ponowne rozumowanie, aż do osiągnięcia satysfakcji [3].
Człowiek w procesie
Jedną z wyjątkowych cech LangGraph [1][2] jest możliwość wstrzymania pracy grafu w trakcie jego uruchamiania w celu oczekiwania na osobę. W finansach jest to przydatne, ponieważ przed finalizacją streszczenia może być konieczna weryfikacja zgodności, albo ktoś może chcieć wpłynąć na przebieg badań w trakcie ich realizacji.
# 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
)
Dzięki utrzymywanemu stanowi operator może sprawdzić, co zostało zebrane, dodać instrukcję i kontynuować:
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)
Standardowy AgentExecutor nie może tego zrobić w sposób czysty. Zachowywanie stanu na podstawie punktów kontrolnych sprawia, że interwencja podczas wykonywania staje się funkcją pierwszorzędną, a nie tylko rozwiązaniem tymczasowym.
Rozszerzanie grafu
Struktura grafu upraszcza dodawanie nowych elementów. Dwie kierunki wydają się obiecujące.
Pierwszym z nich jest węzeł oceny ryzyka. Umieszczony po narzędziach, ale przed końcowym streszczeniem, sprawdza wyniki i zaznacza ryzyka kumulatywne:
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}")]}
Drugi to narzędzie porównywania z rówieśnikami. W przypadku indyjskiej spółki z wysoką kapitalizacją rynkową naturalnym pytaniem jest to, czy jest tańsza czy droższa w porównaniu z innymi spółkami z tego samego sektora:
@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
...
Wystarczy zarejestrować to w TOOLS, a agent pobierze to podczas następnego uruchomienia bez żadnych zmian w grafie.
Agent versus łańcuch
Pojęcie „agenta ReAct” jest używane w sposób luźny, więc różnica wymaga precyzji. Łańcuch to stała sekwencja: krok 1, potem 2, potem 3 – bez względu na to, co dał krok 1. Jego skład jest ustalany już na etapie projektowania.
Agent ReAct decyduje, co robić dalej po każdej obserwacji [3]. Załóżmy, że get_stock_financials zwraca firmę nierentowną o spadających przychodach. Dobry agent dostosowuje swoje kolejne działania odpowiednio, być może uruchamiając get_news_sentiment, aby sprawdzić, czy na rynku pojawiają się jakieś pozytywne informacje. Jeśli get_technical_indicators pokazuje bardzo niski wskaźnik RSI u firmy o silnych fundamentach, agent może przejść bezpośrednio do get_analyst_ratings, aby sprawdzić, czy eksperci zauważają tę samą sytuację.
Taki adaptacyjny sekwencjonowanie wynika z cyklu Myśl, Działanie, Obserwacja i jest widoczne w transmisji wyników, ponieważ agent rozważa różne opcje pomiędzy każdym wywołaniem narzędzia, a nie tylko na końcu.
Istnieje też praktyczna korzyść. Nieoczekiwane dane nie powodują awarii systemu. Spółka o małej kapitalizacji bez analizy nie powoduje problemów, gdyż agent zauważa tę lukę i szuka rozwiązań wokół niej. Indeks giełdowy w trakcie sezonu raportowania z intensywnymi wiadomościami wykazuje większą niepewność w swoim wyniku oceny przekonania. To właśnie ta elastyczność jest powodem, dla którego warto zaakceptować większą złożoność grafu zamiast prostego łańcucha.
Jakie zadanie pełni każdy plik
state.pyzawiera TypedDictAgentStatewraz z reduktoremadd_messages[1]. Usunięcie tego reduktora powoduje, że agent traci kontekst na każdym kroku i musi powtarzać pracę.
tools.py definiuje cztery narzędzia badawcze oparte na yfinance [4] oraz pandas-ta [5]. Ich dokumentacje są tym, co czyta model językowy przy wyborze narzędzia, więc ich jakość bezpośrednio wpływa na dokładność selekcji.agent.py łączy te narzędzia z GPT-4o i stosuje ustrukturyzowany prompt systemowy, który określa format wyjściowego wyniku.graph.py tworzy StateGraph [1][2] przy użyciu tools_condition w celu dynamicznego kierowania procesem ReAct. Liczba iteracji nie jest ustalona z góry, a proces kończy się w momencie, gdy agent oznajmi, że skończył.main.py przekazuje wyniki wykonywania w czasie rzeczywistym, dzięki czemu cały cykl Thought, Act, Observe jest widoczny na żywo.Rozszerzenie typu human-in-the-loop pokazuje, dlaczego LangGraph [2] jest lepszym wyborem niż ręcznie zaprojektowany cykl ReAct. Zachowywanie stanu, przerwy poprzez punkty kontrolne oraz interwencje w trakcie działania to funkcje produkcyjne, które mają znaczenie, gdy chodzi o rzeczywiste portfele.
Referencje
[1] Dokumentacja LangGraph dotycząca StateGraph, ToolNode, tools_condition oraz add_messages, dostępna pod adresem docs.langchain.com/langgraph.
[2] Repozytorium LangGraph, dostępne pod adresem github.com/langchain-ai/langgraph.
[3] Artykuł o ReAct dotyczący łączenia rozumowania i działania w modelach językowych (Yao i wsp., 2023), dostępny pod adresem arxiv.org/abs/2210.03629.
[4] yfinance – narzędzie do pobierania danych z Yahoo Finance, dostępne pod adresem pypi.org/project/yfinance.
[5] pandas-ta – biblioteka do analizy technicznej dla pandas, dostępna pod adresem pypi.org/project/pandas-ta.
[6] Tavily – API do wyszukiwania dostosowane do zastosowań z modelami językowymi, dostępne pod adresem tavily.com.
Pozycje pokrewne
- Praktyczne notatki: Budowanie samokorygującego agenta AI z Self-RAG i LangGraph — Szczegółowy przewodnik po praktycznych notatkach: Budowanie samokorygującego agenta AI z Self-RAG i LangGraph: umowy, sprawdzenia oraz gotowe fragmenty kodu dla zespołów wdrażających ten model.
- Tworzenie lokalnego agenta do badań w LM Studio przy użyciu czterech serwerów MCP — Połączenie serwerów filesystem, GitHub, SearXNG i Camofox MCP w jednej rozmowie w LM Studio, uproścenie schematu narzędzia, dodanie promptu dotyczącego planu badań oraz weryfikacja poprawności za pomocą demonstracji.