fix: issues with communication
This commit is contained in:
Kevin Bruton
parent
5a66be5dac
commit
94eb6c3b68
Binary file not shown.
|
|
@ -0,0 +1,61 @@
|
||||||
|
#!/usr/bin/env python3
|
||||||
|
"""
|
||||||
|
Test script to verify LangGraph streaming behavior
|
||||||
|
"""
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
from datetime import date
|
||||||
|
from dotenv import load_dotenv
|
||||||
|
|
||||||
|
# Add the project root to the path
|
||||||
|
sys.path.insert(0, '/Users/kevin.bruton/repo2/TradingAgents')
|
||||||
|
|
||||||
|
# Load environment variables
|
||||||
|
load_dotenv()
|
||||||
|
|
||||||
|
def test_callback(state):
|
||||||
|
"""Test callback to understand state structure"""
|
||||||
|
print(f"\n🔍 CALLBACK RECEIVED:")
|
||||||
|
print(f" Type: {type(state)}")
|
||||||
|
print(f" Keys: {list(state.keys()) if isinstance(state, dict) else 'Not a dict'}")
|
||||||
|
if isinstance(state, dict):
|
||||||
|
for key, value in state.items():
|
||||||
|
if key not in ["__end__", "messages"]:
|
||||||
|
print(f" {key}: {type(value)} - {'Has content' if value else 'Empty'}")
|
||||||
|
|
||||||
|
def main():
|
||||||
|
"""Test the TradingAgentsGraph streaming"""
|
||||||
|
try:
|
||||||
|
from tradingagents.graph.trading_graph import TradingAgentsGraph
|
||||||
|
from tradingagents.default_config import DEFAULT_CONFIG
|
||||||
|
|
||||||
|
print("🚀 Testing TradingAgentsGraph streaming...")
|
||||||
|
|
||||||
|
# Create a minimal config for testing
|
||||||
|
config = DEFAULT_CONFIG.copy()
|
||||||
|
config["llm_provider"] = "openai"
|
||||||
|
config["quick_think_llm"] = "gpt-3.5-turbo"
|
||||||
|
config["deep_think_llm"] = "gpt-4"
|
||||||
|
|
||||||
|
# Create graph with debug mode
|
||||||
|
graph = TradingAgentsGraph(config=config, debug=True)
|
||||||
|
|
||||||
|
print("📊 Starting propagation with callback...")
|
||||||
|
|
||||||
|
# Test with a simple company
|
||||||
|
final_state, signal = graph.propagate(
|
||||||
|
company_name="AAPL",
|
||||||
|
trade_date=str(date.today()),
|
||||||
|
on_step_callback=test_callback
|
||||||
|
)
|
||||||
|
|
||||||
|
print(f"\n✅ Propagation completed!")
|
||||||
|
print(f" Final signal: {signal}")
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
import traceback
|
||||||
|
print(f"❌ Error: {e}")
|
||||||
|
print(traceback.format_exc())
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
214
webapp/main.py
214
webapp/main.py
|
|
@ -49,117 +49,129 @@ jinja_env = jinja2.Environment(loader=jinja2.FileSystemLoader(template_dir))
|
||||||
|
|
||||||
def update_execution_state(state: Dict[str, Any]):
|
def update_execution_state(state: Dict[str, Any]):
|
||||||
"""Callback function to update the app_state based on LangGraph's state."""
|
"""Callback function to update the app_state based on LangGraph's state."""
|
||||||
|
print(f"📡 Callback received state keys: {list(state.keys())}")
|
||||||
|
|
||||||
with app_state_lock:
|
with app_state_lock:
|
||||||
# Check if we're still in initialization phase and need to transition to actual execution
|
# Initialize the root node if needed
|
||||||
if (app_state["execution_tree"] and
|
if not app_state["execution_tree"] or (
|
||||||
len(app_state["execution_tree"]) == 1 and
|
len(app_state["execution_tree"]) == 1 and
|
||||||
app_state["execution_tree"][0]["id"] == "initialization"):
|
app_state["execution_tree"][0]["id"] == "initialization"
|
||||||
# Replace initialization message with the main execution tree
|
):
|
||||||
app_state["execution_tree"] = [{
|
app_state["execution_tree"] = [{
|
||||||
"id": "root",
|
|
||||||
"name": f"Trading Analysis for {app_state['company_symbol']}",
|
|
||||||
"status": "in_progress",
|
|
||||||
"content": f"Analyzing {app_state['company_symbol']} using multiple trading agents\n\nThe trading analysis pipeline has been successfully initialized and agents are now executing their tasks.",
|
|
||||||
"children": [],
|
|
||||||
"timestamp": time.time()
|
|
||||||
}]
|
|
||||||
elif not app_state["execution_tree"]:
|
|
||||||
# Fallback: Initialize the root node if it doesn't exist
|
|
||||||
app_state["execution_tree"].append({
|
|
||||||
"id": "root",
|
"id": "root",
|
||||||
"name": f"Trading Analysis for {app_state['company_symbol']}",
|
"name": f"Trading Analysis for {app_state['company_symbol']}",
|
||||||
"status": "in_progress",
|
"status": "in_progress",
|
||||||
"content": f"Analyzing {app_state['company_symbol']} using multiple trading agents",
|
"content": f"Analyzing {app_state['company_symbol']} using multiple trading agents",
|
||||||
"children": [],
|
"children": [],
|
||||||
"timestamp": time.time()
|
"timestamp": time.time()
|
||||||
})
|
}]
|
||||||
|
|
||||||
root_node = app_state["execution_tree"][0]
|
root_node = app_state["execution_tree"][0]
|
||||||
|
|
||||||
# Define the expected phases and their order
|
# Map LangGraph node names to user-friendly display info
|
||||||
phase_map = {
|
node_mapping = {
|
||||||
"market_analyst": {"name": "Market Analysis", "phase": "data_collection"},
|
"Market Analyst": {"name": "📈 Market Analysis", "phase": "data_collection"},
|
||||||
"social_analyst": {"name": "Social Media Analysis", "phase": "data_collection"},
|
"Social Analyst": {"name": "📱 Social Media Analysis", "phase": "data_collection"},
|
||||||
"news_analyst": {"name": "News Analysis", "phase": "data_collection"},
|
"News Analyst": {"name": "📰 News Analysis", "phase": "data_collection"},
|
||||||
"fundamentals_analyst": {"name": "Fundamental Analysis", "phase": "data_collection"},
|
"Fundamentals Analyst": {"name": "📊 Fundamental Analysis", "phase": "data_collection"},
|
||||||
"bull_researcher": {"name": "Bull Case Research", "phase": "research"},
|
"Bull Researcher": {"name": "🐂 Bull Case Research", "phase": "research"},
|
||||||
"bear_researcher": {"name": "Bear Case Research", "phase": "research"},
|
"Bear Researcher": {"name": "🐻 Bear Case Research", "phase": "research"},
|
||||||
"research_manager": {"name": "Research Synthesis", "phase": "research"},
|
"Research Manager": {"name": "🔍 Research Synthesis", "phase": "research"},
|
||||||
"trade_planner": {"name": "Trade Planning", "phase": "planning"},
|
"Trade Planner": {"name": "📋 Trade Planning", "phase": "planning"},
|
||||||
"trader": {"name": "Trade Execution", "phase": "execution"},
|
"Trader": {"name": "⚡ Trade Execution", "phase": "execution"},
|
||||||
"risky_analyst": {"name": "Risk Assessment (Aggressive)", "phase": "risk_analysis"},
|
"Risky Analyst": {"name": "🚨 Risk Assessment (Aggressive)", "phase": "risk_analysis"},
|
||||||
"neutral_analyst": {"name": "Risk Assessment (Neutral)", "phase": "risk_analysis"},
|
"Neutral Analyst": {"name": "⚖️ Risk Assessment (Neutral)", "phase": "risk_analysis"},
|
||||||
"safe_analyst": {"name": "Risk Assessment (Conservative)", "phase": "risk_analysis"},
|
"Safe Analyst": {"name": "🛡️ Risk Assessment (Conservative)", "phase": "risk_analysis"},
|
||||||
"risk_judge": {"name": "Final Risk Evaluation", "phase": "risk_analysis"}
|
"Risk Judge": {"name": "⚠️ Final Risk Evaluation", "phase": "risk_analysis"}
|
||||||
}
|
}
|
||||||
|
|
||||||
# Find which agent just completed by examining the state
|
phase_names = {
|
||||||
for key, value in state.items():
|
"data_collection": "📊 Data Collection",
|
||||||
if key in ["__end__", "messages"]:
|
"research": "🔍 Research & Analysis",
|
||||||
continue
|
"planning": "📋 Trade Planning",
|
||||||
|
"execution": "⚡ Trade Execution",
|
||||||
# Map the key to a more user-friendly name
|
"risk_analysis": "⚠️ Risk Management"
|
||||||
agent_key = key.lower().replace(" ", "_").replace("_agent", "").replace("_node", "")
|
}
|
||||||
if agent_key in phase_map:
|
|
||||||
phase_info = phase_map[agent_key]
|
# The state dict contains the current state of all nodes
|
||||||
|
# We need to determine what has actually been executed
|
||||||
# Find or create phase category
|
current_step = None
|
||||||
phase_category = None
|
|
||||||
for child in root_node["children"]:
|
# LangGraph streams the full state each time, so we need to detect what's new
|
||||||
if child["id"] == phase_info["phase"]:
|
# Look for populated report fields to determine what has been completed
|
||||||
phase_category = child
|
if state.get("market_report") and not any(child.get("id") == "data_collection_market" for phase in root_node["children"] for child in phase.get("children", [])):
|
||||||
break
|
current_step = "Market Analyst"
|
||||||
|
elif state.get("sentiment_report") and not any(child.get("id") == "data_collection_social" for phase in root_node["children"] for child in phase.get("children", [])):
|
||||||
if not phase_category:
|
current_step = "Social Analyst"
|
||||||
phase_names = {
|
elif state.get("news_report") and not any(child.get("id") == "data_collection_news" for phase in root_node["children"] for child in phase.get("children", [])):
|
||||||
"data_collection": "📊 Data Collection",
|
current_step = "News Analyst"
|
||||||
"research": "🔍 Research & Analysis",
|
elif state.get("fundamentals_report") and not any(child.get("id") == "data_collection_fundamentals" for phase in root_node["children"] for child in phase.get("children", [])):
|
||||||
"planning": "📋 Trade Planning",
|
current_step = "Fundamentals Analyst"
|
||||||
"execution": "⚡ Trade Execution",
|
elif state.get("investment_debate_state", {}).get("bull_history") and not any(child.get("id") == "research_bull" for phase in root_node["children"] for child in phase.get("children", [])):
|
||||||
"risk_analysis": "⚠️ Risk Management"
|
current_step = "Bull Researcher"
|
||||||
}
|
elif state.get("investment_debate_state", {}).get("bear_history") and not any(child.get("id") == "research_bear" for phase in root_node["children"] for child in phase.get("children", [])):
|
||||||
|
current_step = "Bear Researcher"
|
||||||
phase_category = {
|
elif state.get("investment_debate_state", {}).get("judge_decision") and not any(child.get("id") == "research_manager" for phase in root_node["children"] for child in phase.get("children", [])):
|
||||||
"id": phase_info["phase"],
|
current_step = "Research Manager"
|
||||||
"name": phase_names.get(phase_info["phase"], phase_info["phase"]),
|
elif state.get("trader_investment_plan") and not any(child.get("id") == "planning_trade_planner" for phase in root_node["children"] for child in phase.get("children", [])):
|
||||||
"status": "in_progress",
|
current_step = "Trade Planner"
|
||||||
"content": f"Phase: {phase_names.get(phase_info['phase'], phase_info['phase'])}",
|
elif state.get("investment_plan") and not any(child.get("id") == "execution_trader" for phase in root_node["children"] for child in phase.get("children", [])):
|
||||||
"children": [],
|
current_step = "Trader"
|
||||||
"timestamp": time.time()
|
elif state.get("risk_debate_state", {}).get("risky_history") and not any(child.get("id") == "risk_analysis_risky" for phase in root_node["children"] for child in phase.get("children", [])):
|
||||||
}
|
current_step = "Risky Analyst"
|
||||||
root_node["children"].append(phase_category)
|
elif state.get("risk_debate_state", {}).get("neutral_history") and not any(child.get("id") == "risk_analysis_neutral" for phase in root_node["children"] for child in phase.get("children", [])):
|
||||||
|
current_step = "Neutral Analyst"
|
||||||
# Check if this specific step already exists
|
elif state.get("risk_debate_state", {}).get("safe_history") and not any(child.get("id") == "risk_analysis_safe" for phase in root_node["children"] for child in phase.get("children", [])):
|
||||||
step_exists = False
|
current_step = "Safe Analyst"
|
||||||
for step in phase_category["children"]:
|
elif state.get("final_trade_decision") and not any(child.get("id") == "risk_analysis_risk_judge" for phase in root_node["children"] for child in phase.get("children", [])):
|
||||||
if step["name"] == phase_info["name"]:
|
current_step = "Risk Judge"
|
||||||
step["status"] = "completed"
|
|
||||||
step["content"] = str(value) if value else "Completed successfully"
|
if current_step and current_step in node_mapping:
|
||||||
step_exists = True
|
print(f"🎯 Processing step: {current_step}")
|
||||||
break
|
node_info = node_mapping[current_step]
|
||||||
|
phase_id = node_info["phase"]
|
||||||
if not step_exists:
|
|
||||||
# Add new step
|
# Find or create phase category
|
||||||
new_step = {
|
phase_category = None
|
||||||
"id": f"{phase_info['phase']}_{agent_key}_{len(phase_category['children'])}",
|
for child in root_node["children"]:
|
||||||
"name": phase_info["name"],
|
if child["id"] == phase_id:
|
||||||
"status": "completed",
|
phase_category = child
|
||||||
"content": str(value) if value else "Completed successfully",
|
break
|
||||||
"children": [],
|
|
||||||
"timestamp": time.time()
|
if not phase_category:
|
||||||
}
|
phase_category = {
|
||||||
phase_category["children"].append(new_step)
|
"id": phase_id,
|
||||||
|
"name": phase_names.get(phase_id, phase_id),
|
||||||
# Check if phase is complete (simple heuristic)
|
"status": "in_progress",
|
||||||
completed_steps = sum(1 for step in phase_category["children"] if step["status"] == "completed")
|
"content": f"Phase: {phase_names.get(phase_id, phase_id)}",
|
||||||
if completed_steps >= len(phase_category["children"]):
|
"children": [],
|
||||||
phase_category["status"] = "completed"
|
"timestamp": time.time()
|
||||||
|
}
|
||||||
# Update overall progress based on completed phases
|
root_node["children"].append(phase_category)
|
||||||
total_phases = len([p for p in phase_map.values()])
|
|
||||||
completed_agents = sum(len(child["children"]) for child in root_node["children"]
|
# Add new step
|
||||||
if child.get("children"))
|
step_id = f"{phase_id}_{current_step.lower().replace(' ', '_')}"
|
||||||
app_state["overall_progress"] = min(100, int((completed_agents / max(total_phases, 1)) * 100))
|
new_step = {
|
||||||
|
"id": step_id,
|
||||||
|
"name": node_info["name"],
|
||||||
|
"status": "completed",
|
||||||
|
"content": f"✅ {node_info['name']} completed successfully",
|
||||||
|
"children": [],
|
||||||
|
"timestamp": time.time()
|
||||||
|
}
|
||||||
|
phase_category["children"].append(new_step)
|
||||||
|
|
||||||
|
# Mark phase as completed if it has steps
|
||||||
|
phase_category["status"] = "completed"
|
||||||
|
|
||||||
|
# Update overall progress
|
||||||
|
total_steps = len(node_mapping)
|
||||||
|
completed_steps = sum(len(child["children"]) for child in root_node["children"])
|
||||||
|
app_state["overall_progress"] = min(100, int((completed_steps / max(total_steps, 1)) * 100))
|
||||||
|
|
||||||
|
print(f"📊 Progress updated: {app_state['overall_progress']}% ({completed_steps}/{total_steps} steps)")
|
||||||
|
else:
|
||||||
|
print(f"⏳ No new step detected or step already processed")
|
||||||
|
|
||||||
def run_trading_process(company_symbol: str, config: Dict[str, Any]):
|
def run_trading_process(company_symbol: str, config: Dict[str, Any]):
|
||||||
"""Runs the TradingAgentsGraph in a separate thread."""
|
"""Runs the TradingAgentsGraph in a separate thread."""
|
||||||
|
|
@ -197,10 +209,14 @@ def run_trading_process(company_symbol: str, config: Dict[str, Any]):
|
||||||
else: # openai
|
else: # openai
|
||||||
custom_config["backend_url"] = "https://api.openai.com/v1"
|
custom_config["backend_url"] = "https://api.openai.com/v1"
|
||||||
|
|
||||||
|
print(f"🚀 Initializing TradingAgentsGraph for {company_symbol}")
|
||||||
graph = TradingAgentsGraph(config=custom_config)
|
graph = TradingAgentsGraph(config=custom_config)
|
||||||
analysis_date = config["analysis_date"] # Use user-selected date
|
analysis_date = config["analysis_date"] # Use user-selected date
|
||||||
|
print(f"🔄 Starting propagation for {company_symbol} on {analysis_date}")
|
||||||
|
|
||||||
# The propagate method now accepts the callback and trade_date
|
# The propagate method now accepts the callback and trade_date
|
||||||
final_state = graph.propagate(company_symbol, trade_date=analysis_date, on_step_callback=update_execution_state)
|
final_state, processed_signal = graph.propagate(company_symbol, trade_date=analysis_date, on_step_callback=update_execution_state)
|
||||||
|
print(f"✅ Propagation completed for {company_symbol}")
|
||||||
|
|
||||||
with app_state_lock:
|
with app_state_lock:
|
||||||
app_state["overall_status"] = "completed"
|
app_state["overall_status"] = "completed"
|
||||||
|
|
@ -208,7 +224,7 @@ def run_trading_process(company_symbol: str, config: Dict[str, Any]):
|
||||||
# Update the root node status to completed
|
# Update the root node status to completed
|
||||||
if app_state["execution_tree"]:
|
if app_state["execution_tree"]:
|
||||||
app_state["execution_tree"][0]["status"] = "completed"
|
app_state["execution_tree"][0]["status"] = "completed"
|
||||||
app_state["execution_tree"][0]["content"] = str(final_state)
|
app_state["execution_tree"][0]["content"] = f"✅ Analysis completed successfully!\n\nFinal Decision: {processed_signal}\n\nFull State: {str(final_state)}"
|
||||||
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
import traceback
|
import traceback
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue