Создание продвинутого мультиагентного исследовательского пайплайна с LangGraph и Gemini AI

!pip install -q langgraph langchain-google-genai langchain-core
 
import os
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
import operator
 
os.environ["GOOGLE_API_KEY"] = "Use Your Own 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(model="gemini-1.5-flash", temperature=0.7)

def research_agent(state: AgentState) -> AgentState:
   messages = state["messages"]
   last_message = messages[-1].content
  
   search_results = simulate_web_search(last_message)
  
   prompt = f"""You are a research agent. Based on the query: \"{last_message}\"\n\n   Here are the search results: {search_results}\n\n   Conduct thorough research and gather relevant information. Provide structured findings with:\n   1. Key facts and data points\n   2. Current trends and developments \n   3. Expert opinions and insights\n   4. Relevant statistics\n\n   Be comprehensive and analytical in your research summary."""
  
   response = llm.invoke([HumanMessage(content=prompt)])
  
   research_data = {
       "topic": last_message,
       "findings": response.content,
       "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": "analysis",
       "research_data": research_data,
       "analysis_complete": False,
       "final_report": ""
   }

def analysis_agent(state: AgentState) -> AgentState:
   research_data = state["research_data"]
  
   analysis_results = simulate_data_analysis(research_data.get('findings', ''))
  
   prompt = f"""You are an analysis agent. Analyze this research data in depth:\n\n   Topic: {research_data.get('topic', 'Unknown')}\n   Research Findings: {research_data.get('findings', 'No findings')}\n   Analysis Results: {analysis_results}\n\n   Provide deep insights including:\n   1. Pattern identification and trend analysis\n   2. Comparative analysis with industry standards\n   3. Risk assessment and opportunities \n   4. Strategic implications\n   5. Actionable recommendations with priority levels\n\n   Be analytical and provide 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:
   research_data = state["research_data"]
  
   analysis_message = None
   for msg in reversed(state["messages"]):
       if isinstance(msg, AIMessage) and "Analysis completed:" in msg.content:
           analysis_message = msg.content.replace("Analysis completed: ", "")
           break
  
   prompt = f"""You are a professional report generation agent. Create a comprehensive executive report based on:\n\n    Research Topic: {research_data.get('topic')}\n    Research Findings: {research_data.get('findings')}\n    Analysis Results: {analysis_message or 'Analysis pending'}\n\n   Generate a well-structured, professional report with these sections:\n\n   ## EXECUTIVE SUMMARY  \n   ## KEY RESEARCH FINDINGS \n   [Detail the most important discoveries and data points]\n\n   ## ANALYTICAL INSIGHTS\n   [Present deep analysis, patterns, and trends identified]\n\n   ## STRATEGIC RECOMMENDATIONS\n   [Provide actionable recommendations with priority levels]\n\n   ## RISK ASSESSMENT & OPPORTUNITIES\n   [Identify potential risks and opportunities]\n\n   ## CONCLUSION & NEXT STEPS\n   [Summarize and suggest follow-up actions]\n\n   Make the report professional, data-driven, and actionable."""
  
   response = llm.invoke([HumanMessage(content=prompt)])
  
   return {
       "messages": state["messages"] + [AIMessage(content=f" FINAL REPORT GENERATED:\n\n{response.content}")],
       "current_agent": "complete",
       "research_data": state["research_data"],
       "analysis_complete": True,
       "final_report": response.content
   }

def should_continue(state: AgentState) -> str:
   current_agent = state.get("current_agent", "research")
  
   if current_agent == "research":
       return "analysis"
   elif current_agent == "analysis":
       return "report"
   elif current_agent == "report":
       return END
   else:
       return END
 
 
workflow = StateGraph(AgentState)
 
workflow.add_node("research", research_agent)
workflow.add_node("analysis", analysis_agent)
workflow.add_node("report", report_agent)
 
workflow.add_conditional_edges(
   "research",
   should_continue,
   {"analysis": "analysis", END: END}
)
 
workflow.add_conditional_edges(
   "analysis",
   should_continue,
   {"report": "report", END: END}
)
 
workflow.add_conditional_edges(
   "report",
   should_continue,
   {END: END}
)
 
workflow.set_entry_point("research")
 
app = workflow.compile()
 
 
def run_research_assistant(query: str):
   initial_state = {
       "messages": [HumanMessage(content=query)],
       "current_agent": "research",
       "research_data": {},
       "analysis_complete": False,
       "final_report": ""
   }
  
   print(f" Starting Multi-Agent Research on: '{query}'")
   print("=" * 60)
  
   current_state = initial_state
  
   print(" Research Agent: Gathering information...")
   current_state = research_agent(current_state)
   print(" Research phase completed!\n")
  
   print(" Analysis Agent: Analyzing findings...")
   current_state = analysis_agent(current_state)
   print(" Analysis phase completed!\n")
  
   print(" Report Agent: Generating comprehensive report...")
   final_state = report_agent(current_state)
   print(" Report generation completed!\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(" Advanced LangGraph Multi-Agent System Ready!")
   print(" Remember to set your GOOGLE_API_KEY!")
  
   example_queries = [
       "Impact of renewable energy on global markets",
       "Future of remote work post-pandemic"
   ]
  
   print(f"\n Example queries you can try:")
   for i, query in enumerate(example_queries, 1):
       print(f"  {i}. {query}")
  
   print(f"\n Usage: run_research_assistant('Your research question here')")
  
   result = run_research_assistant("What are emerging trends in sustainable technology?")