LangGraph Multi-Agent Systems: A Comprehensive Guide

1from langchain_core.runnables import chain
2from langgraph.graph import StateGraph, END
3from langchain_core.messages import BaseMessage
4
5# Example: Define a simple agent that echoes the input
6def agent_function(messages: list[BaseMessage]) -> str:
7    return messages[-1].content #Echo last content
8
9# Define the state
10from typing import TypedDict, List, Dict, Any
11
12class GraphState(TypedDict):
13    messages: List[BaseMessage]
14    agent_outcome: Dict[str, Any]
15
16# Create a new graph
17graph = StateGraph(GraphState)
18
19# Add nodes (agents)
20graph.add_node("agent_1", agent_function)
21
22graph.set_entry_point("agent_1")
23
24# Add edges (connections between agents)
25graph.add_edge("agent_1", END)
26
27# Compile the graph
28app = graph.compile()
29

1from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
2from langchain_core.runnables import chain
3from langchain_core.tools import tool
4from langchain_core.messages import BaseMessage, SystemMessage
5
6from langgraph.graph import MessageGraph, END
7
8#Define tools for agents
9@tool
10def search(query: str) -> str:
11    """Search the web for relevant information"""
12    return f"Results for query '{query}'"
13
14# Create a prompt template
15prompt = ChatPromptTemplate.from_messages([
16    ("system", """You are a helpful assistant.
17    Use tools if necessary. 
18    When asked a question, respond with the right information after using search.
19    ""),
20    MessagesPlaceholder(variable_name="messages"),
21])
22
23# Create a function to return a tool-calling chain
24def agent_chain():
25    return prompt | llm | parser
26
27# Initialize the graph
28graph = MessageGraph()
29
30# Add a node (agent)
31graph.add_node("agent", agent_chain())
32
33# Define entry point
34graph.set_entry_point(SystemMessage(content="Begin!"))
35
36# Add edges
37graph.add_edge("agent", END)
38
39# Compile
40app = graph.compile()
41

1import sqlite3
2
3# Initialize a database connection
4conn = sqlite3.connect('shared_state.db')
5cursor = conn.cursor()
6
7# Create a table to store the shared state
8cursor.execute('''
9    CREATE TABLE IF NOT EXISTS shared_data (
10        key TEXT PRIMARY KEY,
11        value TEXT
12    )
13''')
14conn.commit()
15
16# Function to read from shared state
17def get_shared_state(key: str) -> str:
18    cursor.execute('SELECT value FROM shared_data WHERE key = ?', (key,))
19    result = cursor.fetchone()
20    return result[0] if result else None
21
22# Function to write to shared state
23def set_shared_state(key: str, value: str):
24    cursor.execute('INSERT OR REPLACE INTO shared_data (key, value) VALUES (?, ?)', (key, value))
25    conn.commit()
26
27# Example Usage from agents
28# agent_1 gets a value
29# value = get_shared_state("important_value")
30
31# agent_2 sets a value
32# set_shared_state("important_value", "new_value")
33
34# Clean up
35# conn.close()
36