A Coding Implementation to Superior LangGraph Multi-Agent Analysis Pipeline for Automated Insights Technology

!pip set up -q langgraph langchain-google-genai langchain-core


import os
from typing import TypedDict, Annotated, Listing, Dict, Any
from langgraph.graph import StateGraph, END
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
import operator
import json




os.environ["GOOGLE_API_KEY"] = "Use Your Personal API Key"


class AgentState(TypedDict):
   messages: Annotated[List[BaseMessage], operator.add]
   current_agent: str
   research_data: dict
   analysis_complete: bool
   final_report: str


llm = ChatGoogleGenerativeAI(mannequin="gemini-1.5-flash", temperature=0.7)

def simulate_web_search(question: str) -> str:
   """Simulated net search - exchange with actual API in manufacturing"""
   return f"Search outcomes for '{question}': Discovered related details about {question} together with current developments, knowledgeable opinions, and statistical information."


def simulate_data_analysis(information: str) -> str:
   """Simulated information evaluation instrument"""
   return f"Evaluation full: Key insights from the info embody rising traits, statistical patterns, and actionable suggestions."


def research_agent(state: AgentState) -> AgentState:
   """Agent that researches a given subject"""
   messages = state["messages"]
   last_message = messages[-1].content material
  
   search_results = simulate_web_search(last_message)
  
   immediate = f"""You're a analysis agent. Based mostly on the question: "{last_message}"
  
   Listed here are the search outcomes: {search_results}
  
   Conduct thorough analysis and collect related info. Present structured findings with:
   1. Key information and information factors
   2. Present traits and developments 
   3. Skilled opinions and insights
   4. Related statistics
  
   Be complete and analytical in your analysis abstract."""
  
   response = llm.invoke([HumanMessage(content=prompt)])
  
   research_data = {
       "subject": last_message,
       "findings": response.content material,
       "search_results": search_results,
       "sources": ["academic_papers", "industry_reports", "expert_analyses"],
       "confidence": 0.88,
       "timestamp": "2024-research-session"
   }
  
   return {
       "messages": state["messages"] + [AIMessage(content=f"Research completed on '{last_message}': {response.content}")],
       "current_agent": "evaluation",
       "research_data": research_data,
       "analysis_complete": False,
       "final_report": ""
   }

def analysis_agent(state: AgentState) -> AgentState:
   """Agent that analyzes analysis information and extracts insights"""
   research_data = state["research_data"]
  
   analysis_results = simulate_data_analysis(research_data.get('findings', ''))
  
   immediate = f"""You might be an evaluation agent. Analyze this analysis information in depth:
  
   Matter: {research_data.get('subject', 'Unknown')}
   Analysis Findings: {research_data.get('findings', 'No findings')}
   Evaluation Outcomes: {analysis_results}
  
   Present deep insights together with:
   1. Sample identification and pattern evaluation
   2. Comparative evaluation with trade requirements
   3. Danger evaluation and alternatives 
   4. Strategic implications
   5. Actionable suggestions with precedence ranges
  
   Be analytical and supply evidence-based insights."""
  
   response = llm.invoke([HumanMessage(content=prompt)])
  
   return {
       "messages": state["messages"] + [AIMessage(content=f"Analysis completed: {response.content}")],
       "current_agent": "report",
       "research_data": state["research_data"],
       "analysis_complete": True,
       "final_report": ""
   }




def report_agent(state: AgentState) -> AgentState:
   """Agent that generates ultimate complete reviews"""
   research_data = state["research_data"]
  
   analysis_message = None
   for msg in reversed(state["messages"]):
       if isinstance(msg, AIMessage) and "Evaluation accomplished:" in msg.content material:
           analysis_message = msg.content material.exchange("Evaluation accomplished: ", "")
           break
  
   immediate = f"""You're a skilled report technology agent. Create a complete govt report based mostly on:
  
   🔍 Analysis Matter: {research_data.get('subject')}
   📊 Analysis Findings: {research_data.get('findings')}
   🧠 Evaluation Outcomes: {analysis_message or 'Evaluation pending'}
  
   Generate a well-structured, skilled report with these sections:
  
   ## EXECUTIVE SUMMARY  
   ## KEY RESEARCH FINDINGS 
   [Detail the most important discoveries and data points]
  
   ## ANALYTICAL INSIGHTS
   [Present deep analysis, patterns, and trends identified]
  
   ## STRATEGIC RECOMMENDATIONS
   [Provide actionable recommendations with priority levels]
  
   ## RISK ASSESSMENT & OPPORTUNITIES
   [Identify potential risks and opportunities]
  
   ## CONCLUSION & NEXT STEPS
   [Summarize and suggest follow-up actions]
  
   Make the report skilled, data-driven, and actionable."""
  
   response = llm.invoke([HumanMessage(content=prompt)])
  
   return {
       "messages": state["messages"] + [AIMessage(content=f"📄 FINAL REPORT GENERATED:nn{response.content}")],
       "current_agent": "full",
       "research_data": state["research_data"],
       "analysis_complete": True,
       "final_report": response.content material
   }

def should_continue(state: AgentState) -> str:
   """Decide which agent ought to run subsequent based mostly on present state"""
   current_agent = state.get("current_agent", "analysis")
  
   if current_agent == "analysis":
       return "evaluation"
   elif current_agent == "evaluation":
       return "report"
   elif current_agent == "report":
       return END
   else:
       return END


workflow = StateGraph(AgentState)


workflow.add_node("analysis", research_agent)
workflow.add_node("evaluation", analysis_agent)
workflow.add_node("report", report_agent)


workflow.add_conditional_edges(
   "analysis",
   should_continue,
   {"evaluation": "evaluation", END: END}
)


workflow.add_conditional_edges(
   "evaluation",
   should_continue,
   {"report": "report", END: END}
)


workflow.add_conditional_edges(
   "report",
   should_continue,
   {END: END}
)


workflow.set_entry_point("analysis")


app = workflow.compile()


def run_research_assistant(question: str):
   """Run the entire analysis workflow"""
   initial_state = {
       "messages": [HumanMessage(content=query)],
       "current_agent": "analysis",
       "research_data": {},
       "analysis_complete": False,
       "final_report": ""
   }
  
   print(f"🔍 Beginning Multi-Agent Analysis on: '{question}'")
   print("=" * 60)
  
   current_state = initial_state
  
   print("🤖 Analysis Agent: Gathering info...")
   current_state = research_agent(current_state)
   print("✅ Analysis part accomplished!n")
  
   print("🧠 Evaluation Agent: Analyzing findings...")
   current_state = analysis_agent(current_state)
   print("✅ Evaluation part accomplished!n")
  
   print("📊 Report Agent: Producing complete report...")
   final_state = report_agent(current_state)
   print("✅ Report technology accomplished!n")
  
   print("=" * 60)
   print("🎯 MULTI-AGENT WORKFLOW COMPLETED SUCCESSFULLY!")
   print("=" * 60)
  
   final_report = final_state['final_report']
   print(f"n📋 COMPREHENSIVE RESEARCH REPORT:n")
   print(final_report)
  
   return final_state

if __name__ == "__main__":
   print("🚀 Superior LangGraph Multi-Agent System Prepared!")
   print("🔧 Keep in mind to set your GOOGLE_API_KEY!")
  
   example_queries = [
       "Impact of renewable energy on global markets",
       "Future of remote work post-pandemic"
   ]
  
   print(f"n💡 Instance queries you may attempt:")
   for i, question in enumerate(example_queries, 1):
       print(f"  {i}. {question}")
  
   print(f"n🎯 Utilization: run_research_assistant('Your analysis query right here')")
  
   end result = run_research_assistant("What are rising traits in sustainable expertise?")