Erstellung eines Aktienforschungs-Agenten mit ReAct und LangGraph
Ein LangGraph ReAct-Agent nimmt einen Aktienkürzel entgegen, ruft nacheinander die fundamentalen Daten, Nachrichten, technischen Analysen sowie Analysetools auf und verfasst anschließend einen kurzen Bericht, wobei Zwischenpausen eingebaut werden, damit ein Mensch eingreifen kann.
Quellen: Building a Financial Research Agent with ReAct, LangGraph, and LangChain
Die Voruntersuchung eines Aktiensymbols vor Marktöffnung ist eine übliche Aufgabe. Wer indische Aktien wie ICICI, HDFC, Reliance oder HAL verfolgt, kann leicht 40 Minuten damit verbringen, zwischen Screener.in, Moneycontrol und verschiedenen Nachrichtenseiten hin- und herzuspringen, um manuell ein Bild zusammenzustellen. Genau diese Art wiederholender Recherche kann von einem Agenten übernommen werden.
Was folgt, ist eine vollständige, funktionsfähige Umsetzung eines Forschungsagenten auf der Basis des ReAct-Musters – mit echtem Code und ohne Pseudocode. Es wird ein Ticker wie RELIANCE.NS oder AAPL eingegeben. Der Agent bewertet dabei Fundamentaldaten, Nachrichtenstimmungen, technische Indikatoren sowie den Konsens der Analysten und erstellt anschließend einen strukturierten Investitionsbericht.
Überraschenderweise gibt es keinen festen Script. Der Agent entscheidet selbst, was er untersuchen soll und in welcher Reihenfolge – basierend auf seinen Erkenntnissen – und genau diese anpassungsfähige Denkweise ist der eigentliche Kern von ReAct.
Was der Agent tut
Gegeben ein Aktienkürzel arbeitet der Agent eigenständig an den folgenden Aufgaben:
- Auswertung der Grundlagenwerte: Kurs-Gewinn-Verhältnis, Umsatzwachstum, EPS, Verschuldungsgrad und Gewinnmargen.
- Sammlung von kürzlichen Nachrichten sowie Bewertung dieser als bullisch, bearisch oder neutral, inklusive Begründung.
- Berechnung von technischen Indikatoren: RSI, MACD, die 50- und 200-Tage-Durchschnitte sowie der Volumentrend.
- Erfassung des Analystenkonsenses, d.h. die Verteilung der Bewertungen „Kaufen“, „Halten“ und „Verkaufen“.
- Fusion aller Informationen in einen strukturierten Bericht, der mit einer Empfehlung endet.
Die Reihenfolge der Tool-Aufrufe liegt in der Entscheidung des Agenten. Zuerst könnten technische Aspekte berücksichtigt werden, anschließend die grundlegenden Faktoren. Oder der Agent könnte Nachrichten lesen, auf ein wichtiges Unternehmensereignis stoßen und umgehend die Finanzdaten prüfen, um zu sehen, ob die Zahlen die Berichterstattung stützen. Solche Entscheidungen heben einen ReAct-Agenten von einer herkömmlichen Kette ab. Das Gesamtdesign:
User: "Research RELIANCE.NS"
│
▼
┌─────────────┐ tool_call ┌─────────────┐
│ Agent Node │ ──────────────────► │ Tools Node │
│ (LLM) │ ◄────────────────── │ (ToolNode) │
└─────────────┘ observation └─────────────┘
│
│ no more tool calls
▼
Final Investment Brief
Dies ist der ReAct-Zyklus von Reason, Act, Observe, dann wieder Reason, der so lange wiederholt wird, bis das System annimmt, genügend Informationen zu haben [3].
Einrichtung
Installieren Sie zunächst die Abhängigkeiten:
pip install langgraph langchain langchain-openai yfinance pandas-ta \
tavily-python python-dotenv pydantic
Danach folgt eine .env-Datei, die die API-Schlüssel enthält:
OPENAI_API_KEY=your_openai_key
TAVILY_API_KEY=your_tavily_key
Die Projektstruktur ist einfach gestaltet:
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
Agentenzustand
Der Zustand ist das gemeinsame Speichermedium, das durch jeden Knoten des Graphen fließt. In LangGraph [1] liest jeder Knoten darin und schreibt wieder hinein, wodurch es als Arbeitsgedächtnis des Agenten während der gesamten Forschungssitzung fungiert.
# 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
Zwei Aspekte sind besonders hervorzuheben. Das Feld messages verwendet add_messages als Reducer [1], wodurch bei jeder Aktualisierung neue Nachrichten hinzugefügt statt ersetzt werden. Dadurch bleibt die gesamte Historie von Gedanken, Aktionen und Beobachtungen im Laufe des Zyklus erhalten. Ohne diesen Reducer würde jeder Knoten das vorherige Gespräch überschreiben, und der Agent würde vergessen, was er bereits überprüft hat – ein Fehler, der häufig darin besteht, dass dieselbe Funktion immer wieder aufgerufen wird. Die strukturierten Forschungsfelder wie financials und news_sentiment sind einfache Wörterbücher, die während des Laufs der Tools gefüllt und bei späteren Überlegungen herangezogen werden.
Tools
Die Definitionen der Tools können wichtiger sein als die Form des Graphen oder die Systemanweisung. Das LLM liest die Beschreibung jedes Tools, um zu entscheiden, welches Tool und zu welchem Zeitpunkt aufgerufen werden soll. Unklare Beschreibungen führen dazu, dass Tools übersprungen oder in einer ungewöhnlichen Reihenfolge verwendet werden, weshalb sie am besten wie Dokumentation für einen kompetenten Kollegen verfasst werden sollten und nicht wie ein Schema für einen Validierer.
# 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,
]
Die hier verwendete Sentiment-Bewertung ist absichtlich einfach und berücksichtigt lediglich die Häufigkeit von Schlüsselwörtern. In einer Produktionsumgebung würde man stattdessen einen auf LLMs basierenden Sentiment-Analyse-Schritt oder einen fein abgestimmten Klassifikator einsetzen. Um den ReAct-Loop zu demonstrieren, reicht die einfache Version aus und reduziert gleichzeitig die Abhängigkeiten.
Systemprompt und Agent
Der Systemprompt bestimmt, wie der Agent argumentiert: seine Prioritäten, deren Reihenfolge sowie das Format der Endausgabe. Genauigkeit zahlt sich hier aus, da ein unklarer Prompt zu unklaren Schlussfolgerungen führt.
# 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)
Das Diagramm
LangGraph [1][2] hat in diesem Schritt Vorteile gegenüber dem standardmäßigen LangChain AgentExecutor. Das Diagramm ermöglicht eine explizite Steuerung des Schlussfolgerungsprozesses. Jeder Zustandswechsel ist sichtbar, Unterbrechungen zur menschlichen Überprüfung können hinzugefügt werden und neue Knoten können ohne Beeinflussung der bestehenden Logik hinzugefügt werden.
# 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()
Die tools_condition-Funktion [1] erledigt leise viele Aufgaben. Nach jedem Schritt des Agenten prüft sie die letzte Nachricht. Ein tool_call vom LLM wird an den tools-Node weitergeleitet, während reiner Text – also der Fall, in dem das Modell mit dem Schlussfolgern fertig ist und bereit zum Schreiben ist – an END weitergeleitet wird.
Es gibt keine vorgegebene Anzahl an Schritten. Der Loop läuft weiter, bis der Agent zufrieden ist, was bei einem Markt mit starker Nachrichtenflut fünf oder sechs Runden sein kann und bei einem ruhigeren Markt etwa drei Runden. Der Eingangspunkt verbindet alles miteinander:
# 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)
Ausführen
Der Agent wird von der Befehlszeile aus gegen einen der Märkte ausgeführt:
# Indian equity
python main.py RELIANCE.NS
# US equity
python main.py AAPL
Die Stream-Ausgabe sieht so aus:
============================================================
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
...
Der Agent rief alle vier Tools auf, bevor er den Auftrag entwarf, und hörte nicht bei den Grundlagen auf. So funktioniert ReAct richtig: überlegen, handeln, beobachten, erneut überlegen – bis man zufrieden ist [3].
Mensch im Prozess
Eine herausragende Produktionsfunktion von LangGraph [1][2] ist die Pause des Graphen während der Ausführung, um auf einen Menschen zu warten. In der Finanzbranche ist das nützlich, da vor der Endfassung eines Auftrags möglicherweise eine Konformitätsprüfung erforderlich ist oder jemand die Forschungsarbeit mittendrin lenken möchte.
# 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
)
Durch den persistenten Zustand kann der Operator sehen, was bereits gesammelt wurde, eine Anweisung hinzufügen und fortfahren:
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)
Ein standardmäßiger AgentExecutor kann dies nicht sauber umsetzen. Die auf Checkpoints basierende Persistenz macht Eingriffe während der Ausführung zu einer erstklassigen Funktion statt zu einem Workaround.
Erweiterung des Graphen
Durch die graphische Struktur bleiben Erweiterungen einfach. Zwei Ansätze scheinen vielversprechend zu sein.
Der erste ist ein Risikobewertungsknoten. Er wird nach den Tools, aber vor dem abschließenden Bericht platziert und überprüft die Ergebnisse sowie markiert kumulative Risiken:
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}")]}
Das Zweite ist ein Vergleichstool für Peer-Werte. Bei einem indischen Large-Cap-Aktienkonzern stellt sich die offensichtliche Frage, ob er im Vergleich zu seinen Branchenkollegen günstig oder teuer ist:
@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
...
Das Registrieren im TOOLS reicht aus, und der Agent übernimmt es beim nächsten Ausführungsvorgang ohne Änderungen am Graphen.
Agent versus Kette
Der Begriff „ReAct-Agent“ wird oft locker verwendet, weshalb der Unterschied präzisiert werden sollte. Eine Kette ist eine feste Abfolge: Schritt 1, dann 2, dann 3 – unabhängig davon, was bei Schritt 1 herauskommt. Ihre Zusammensetzung wird bereits zur Konzeption festgelegt.
Ein ReAct-Agent entscheidet nach jeder Beobachtung, was als Nächstes getan werden soll [3]. Angenommen, get_stock_financials liefert Informationen über ein unrentables Unternehmen mit sinkenden Einnahmen. Ein guter Agent gewichtet seinen nächsten Aufruf entsprechend, vielleicht indem er get_news_sentiment ausführt, um zu prüfen, ob es im Markt Nachrichten über eine Trendwende gibt. Wenn get_technical_indicators bei einem grundlegend starken Unternehmen einen stark unterverkauften RSI anzeigt, könnte der Agent direkt zu get_analyst_ratings wechseln, um herauszufinden, ob auch Fachleute diese Lücke erkennen.
Diese adaptive Sequenzierung entsteht aus dem Zyklus „Denken, Handeln, Beobachten“ und ist in der gestreamten Ausgabe sichtbar, da der Agent zwischen jedem Toolaufruf nachdenkt und nicht nur am Ende.
Darüber hinaus gibt es praktische Vorteile: Unerwartete Eingaben stören die Funktionalität nicht. Ein Unternehmen mit geringem Börsenwert, das von keinem Analysten beobachtet wird, führt nicht zum Zusammenbruch, da der Agent die Lücke bemerkt und entsprechend handelt. Ein Aktienkurs mitten in der Quartalsberichtszeit mit vielen Nachrichten zeigt eine höhere Unsicherheit im Überzeugungsscore an. Genau diese Flexibilität rechtfertigt die Akzeptanz einer höheren Komplexität der Grafiken im Vergleich zu einer einfachen Kette.
Was jede Datei bewirkt
state.pyenthält den TypedDictAgentStatezusammen mit dem Reduceradd_messages[1]. Ohne diesen Reducer verliert der Agent bei jedem Schritt den Kontext und muss die Arbeit erneut durchführen.
tools.py definiert vier Forschungstools für yfinance [4] und pandas-ta [5]. Ihre Dokumentationsstrings sind es, die das LLM beim Auswahl eines Tools liest; daher beeinflusst deren Qualität direkt die Genauigkeit der Auswahl.agent.py verbindet die Tools mit GPT-4o und wendet einen strukturierten Systemprompt an, der das Ausgabeformat festlegt.graph.py erstellt den StateGraph [1][2] mithilfe von tools_condition für eine dynamische ReAct-Verarbeitung. Es gibt keine feste Iterationsanzahl, und die Ausführung endet, sobald der Agent angibt, dass sie abgeschlossen ist.main.py streamt die Ausführung, sodass der gesamte Thought-, Act- und Observe-Zyklus in Echtzeit angezeigt wird.Die Erweiterung „Human-in-the-Loop“ zeigt, warum LangGraph [2] der bessere Wahl ist als ein manuell implementierter ReAct-Loop. Persistierender Zustand, Unterbrechungen durch Checkpoints sowie Eingriffe während des Laufs sind Produktfunktionen, die bei der Verarbeitung echter Portfolios von Bedeutung sind.
Referenzen
[1] LangGraph-Dokumentation zu StateGraph, ToolNode, tools_condition und add_messages unter docs.langchain.com/langgraph.
[2] Das LangGraph-Repository unter github.com/langchain-ai/langgraph.
[3] Die ReAct-Publikation über die Kombination von Reasoning und Handeln in Sprachmodellen (Yao et al., 2023) unter arxiv.org/abs/2210.03629.
[4] yfinance, ein Downloader von Daten von Yahoo Finance, unter pypi.org/project/yfinance.
[5] pandas-ta, eine Bibliothek für technische Analyse in pandas, unter pypi.org/project/pandas-ta.
[6] Tavily, eine Such-API, die für LLM-Anwendungen optimiert ist, unter tavily.com.