«`html
Understanding the Target Audience
The target audience for the A Coding Implementation to Advanced LangGraph Multi-Agent Research Pipeline for Automated Insights Generation consists primarily of business professionals, data scientists, and researchers who are interested in leveraging AI technologies for enhanced research capabilities. This audience typically includes:
- Data analysts seeking to automate insights generation.
- Business managers looking for efficient research workflows.
- Developers interested in implementing AI-driven solutions.
Common pain points for this audience include:
- Time-consuming manual research processes.
- Difficulty in synthesizing large volumes of data into actionable insights.
- Challenges in integrating various AI tools into cohesive workflows.
Their goals often revolve around:
- Streamlining research processes to save time and resources.
- Improving the accuracy and relevance of insights generated.
- Enhancing decision-making capabilities through data-driven reports.
Interests include:
- Latest advancements in AI and machine learning.
- Best practices for data analysis and reporting.
- Tools and frameworks that facilitate automation in research.
In terms of communication preferences, this audience typically favors:
- Clear, concise technical documentation.
- Hands-on tutorials and code examples.
- Interactive webinars and workshops for practical learning.
Advanced LangGraph Multi-Agent Research Pipeline
We build an advanced LangGraph multi-agent system that leverages Google’s free-tier Gemini model for end-to-end research workflows. In this tutorial, we start by installing the necessary libraries, LangGraph, LangChain-Google-GenAI, and LangChain-Core, then walk through defining a structured state, simulating research and analysis tools, and wiring up three specialized agents: Research, Analysis, and Report. Along the way, we show how to simulate web searches, perform data analysis, and orchestrate messages between agents to produce a polished executive report.
Installation
!pip install -q langgraph langchain-google-genai langchain-coreSetting Up the Environment
import os
from typing import TypedDict, Annotated, List, 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 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)We install the LangGraph and LangChain-Google-GenAI packages and import the core modules we need to orchestrate our multi-agent workflow. We set our Google API key, define the AgentState TypedDict to structure messages and workflow state, and initialize the Gemini-1.5-Flash model with a 0.7 temperature for balanced responses.
Simulating Web Search and Data Analysis
def simulate_web_search(query: str) -> str:
return f"Search results for '{query}': Found relevant information about {query} including recent developments, expert opinions, and statistical data."
def simulate_data_analysis(data: str) -> str:
return f"Analysis complete: Key insights from the data include emerging trends, statistical patterns, and actionable recommendations."Research Agent Implementation
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}"
Here are the search results: {search_results}
Conduct thorough research and gather relevant information. Provide structured findings with:
1. Key facts and data points
2. Current trends and developments
3. Expert opinions and insights
4. Relevant statistics
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": ""
}We define simulate_web_search and simulate_data_analysis as placeholder tools that mock retrieving and analyzing information, then implement research_agent to invoke these simulations, prompt Gemini for a structured research summary, and update our workflow state with the findings.
Analysis Agent Implementation
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:
Topic: {research_data.get('topic', 'Unknown')}
Research Findings: {research_data.get('findings', 'No findings')}
Analysis Results: {analysis_results}
Provide deep insights including:
1. Pattern identification and trend analysis
2. Comparative analysis with industry standards
3. Risk assessment and opportunities
4. Strategic implications
5. Actionable recommendations with priority levels
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": ""
}Report Agent Implementation
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:
Research Topic: {research_data.get('topic')}
Research Findings: {research_data.get('findings')}
Analysis Results: {analysis_message or 'Analysis pending'}
Generate a well-structured, professional 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 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
}Workflow Management
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_stateWe construct a StateGraph, add our three agents as nodes with conditional edges dictated by should_continue, set the entry point to “research,” and compile the graph into an executable workflow. We then define run_research_assistant() to initialize the state, sequentially invoke each agent, research, analysis, and report, print status updates, and return the final report.
Conclusion
In conclusion, this modular setup empowers rapid prototyping of complex workflows. Each agent encapsulates a distinct phase of intelligence gathering, interpretation, and delivery, allowing for the integration of real APIs or the extension of the pipeline with new tools as needs evolve. Experimentation with custom tools, adjustments to the state structure, and exploration of alternate LLMs are encouraged. This framework is designed to grow with research and product goals, continually refining agents’ prompts and capabilities to ensure robustness and adaptability across domains.
Check out the Full Codes. Feel free to check out our GitHub Page for Tutorials, Codes, and Notebooks. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter.
«`