Usamos cookies para mejorar tu experiencia en el sitio
Política de Privacidad
Aceptar
LangGraph - Graph-Based Agent Workflows
LangGraph to framework do budowania złożonych, stanowych aplikacji AI opartych na grafach. Pozwala tworzyć cykle, rozgałęzienia i zaawansowane przepływy sterowania dla agentów!
Czym jest LangGraph?
LangGraph rozszerza możliwości LangChain o:
- State Management - zarządzanie stanem między krokami
- Cycles - możliwość tworzenia pętli (w przeciwieństwie do DAG)
- Branching - warunkowe rozgałęzienia przepływu
- Human-in-the-loop - punkty wstrzymania dla interakcji z człowiekiem
1┌─────────────────────────────────────────────────────────────┐
2│ LangGraph Flow │
3├─────────────────────────────────────────────────────────────┤
4│ │
5│ ┌──────────┐ │
6│ │ START │ │
7│ └────┬─────┘ │
8│ │ │
9│ ▼ │
10│ ┌──────────┐ ┌──────────────┐ │
11│ │ Agent │────▶│ Condition │ │
12│ │ Node │ │ (Router) │ │
13│ └──────────┘ └──────┬───────┘ │
14│ ▲ │ │
15│ │ ┌─────────┼─────────┐ │
16│ │ ▼ ▼ ▼ │
17│ │ ┌────────┐ ┌────────┐ ┌────────┐ │
18│ │ │ Tool A │ │ Tool B │ │ END │ │
19│ │ └───┬────┘ └───┬────┘ └────────┘ │
20│ │ │ │ │
21│ └───────┴──────────┘ │
22│ (cycle back) │
23│ │
24└─────────────────────────────────────────────────────────────┘Instalacja i podstawy
1# pip install langgraph langchain-openai
2
3from langgraph.graph import StateGraph, END
4from langgraph.checkpoint.memory import MemorySaver
5from typing import TypedDict, Annotated, Literal
6from langchain_openai import ChatOpenAI
7from langchain_core.messages import HumanMessage, AIMessage
8import operator
9
10# Definicja stanu - kluczowy element LangGraph
11class AgentState(TypedDict):
12 messages: Annotated[list, operator.add] # Lista wiadomości
13 current_step: str
14 iteration_count: int
15
16# Model LLM
17llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)Podstawowy graf agenta
1from langgraph.graph import StateGraph, START, END
2
3# Definicja stanu
4class State(TypedDict):
5 messages: Annotated[list, operator.add]
6 next_action: str
7
8# Funkcje węzłów
9def agent_node(state: State) -> State:
10 """Węzeł agenta - podejmuje decyzje."""
11 messages = state["messages"]
12
13 response = llm.invoke(messages)
14
15 return {
16 "messages": [response],
17 "next_action": "tool" if response.tool_calls else "end"
18 }
19
20def tool_node(state: State) -> State:
21 """Węzeł narzędzi - wykonuje akcje."""
22 last_message = state["messages"][-1]
23
24 # Wykonaj tool calls
25 tool_results = []
26 for tool_call in last_message.tool_calls:
27 result = execute_tool(tool_call)
28 tool_results.append(result)
29
30 return {"messages": tool_results, "next_action": "agent"}
31
32def router(state: State) -> Literal["tool", "end"]:
33 """Router - decyduje o następnym kroku."""
34 return state["next_action"]
35
36# Budowa grafu
37workflow = StateGraph(State)
38
39# Dodaj węzły
40workflow.add_node("agent", agent_node)
41workflow.add_node("tool", tool_node)
42
43# Dodaj krawędzie
44workflow.add_edge(START, "agent")
45workflow.add_conditional_edges(
46 "agent",
47 router,
48 {
49 "tool": "tool",
50 "end": END
51 }
52)
53workflow.add_edge("tool", "agent") # Cykl z powrotem do agenta
54
55# Kompilacja grafu
56app = workflow.compile()
57
58# Uruchomienie
59result = app.invoke({
60 "messages": [HumanMessage(content="Jaka jest pogoda w Serengeti?")],
61 "next_action": ""
62})ReAct Agent z LangGraph
1from langchain_core.tools import tool
2from langgraph.prebuilt import create_react_agent
3
4# Definicja narzędzi
5@tool
6def get_weather(location: str) -> str:
7 """Pobiera pogodę dla lokalizacji Safari."""
8 weather_data = {
9 "Serengeti": "28°C, słonecznie",
10 "Masai Mara": "25°C, pochmurnie",
11 "Kruger": "30°C, gorąco"
12 }
13 return weather_data.get(location, f"Brak danych dla {location}")
14
15@tool
16def search_animals(query: str) -> str:
17 """Wyszukuje informacje o zwierzętach Safari."""
18 return f"Informacje o {query}: Fascynujące zwierzę afrykańskiej sawanny!"
19
20@tool
21def calculate_safari_cost(days: int, people: int) -> str:
22 """Oblicza koszt Safari."""
23 cost_per_day = 200
24 total = days * people * cost_per_day
25 return f"Koszt Safari: {total} USD ({days} dni, {people} osoby)"
26
27# Tworzenie agenta ReAct
28tools = [get_weather, search_animals, calculate_safari_cost]
29react_agent = create_react_agent(llm, tools)
30
31# Uruchomienie
32result = react_agent.invoke({
33 "messages": [HumanMessage(content="Oblicz koszt 5-dniowego Safari dla 4 osób")]
34})
35print(result["messages"][-1].content)Multi-Agent z LangGraph
1from langgraph.graph import StateGraph, START, END
2from typing import Literal
3
4class MultiAgentState(TypedDict):
5 messages: Annotated[list, operator.add]
6 current_agent: str
7 task_complete: bool
8
9# Agenci specjaliści
10def researcher_agent(state: MultiAgentState) -> MultiAgentState:
11 """Agent badawczy - zbiera informacje."""
12 system_prompt = "Jesteś badaczem Safari. Zbieraj informacje o zwierzętach i lokalizacjach."
13
14 response = llm.invoke([
15 {"role": "system", "content": system_prompt},
16 *state["messages"]
17 ])
18
19 return {
20 "messages": [AIMessage(content=f"[Researcher]: {response.content}")],
21 "current_agent": "analyzer"
22 }
23
24def analyzer_agent(state: MultiAgentState) -> MultiAgentState:
25 """Agent analityczny - analizuje dane."""
26 system_prompt = "Jesteś analitykiem. Analizuj zebrane informacje i wyciągaj wnioski."
27
28 response = llm.invoke([
29 {"role": "system", "content": system_prompt},
30 *state["messages"]
31 ])
32
33 return {
34 "messages": [AIMessage(content=f"[Analyzer]: {response.content}")],
35 "current_agent": "writer"
36 }
37
38def writer_agent(state: MultiAgentState) -> MultiAgentState:
39 """Agent pisarski - tworzy końcowy raport."""
40 system_prompt = "Jesteś pisarzem. Stwórz angażujący raport na podstawie analizy."
41
42 response = llm.invoke([
43 {"role": "system", "content": system_prompt},
44 *state["messages"]
45 ])
46
47 return {
48 "messages": [AIMessage(content=f"[Writer]: {response.content}")],
49 "current_agent": "end",
50 "task_complete": True
51 }
52
53def router(state: MultiAgentState) -> Literal["analyzer", "writer", "end"]:
54 """Router między agentami."""
55 return state["current_agent"]
56
57# Budowa grafu multi-agent
58multi_agent_graph = StateGraph(MultiAgentState)
59
60multi_agent_graph.add_node("researcher", researcher_agent)
61multi_agent_graph.add_node("analyzer", analyzer_agent)
62multi_agent_graph.add_node("writer", writer_agent)
63
64multi_agent_graph.add_edge(START, "researcher")
65multi_agent_graph.add_conditional_edges(
66 "researcher",
67 router,
68 {"analyzer": "analyzer"}
69)
70multi_agent_graph.add_conditional_edges(
71 "analyzer",
72 router,
73 {"writer": "writer"}
74)
75multi_agent_graph.add_conditional_edges(
76 "writer",
77 router,
78 {"end": END}
79)
80
81multi_agent_app = multi_agent_graph.compile()
82
83# Uruchomienie pipeline
84result = multi_agent_app.invoke({
85 "messages": [HumanMessage(content="Przygotuj raport o lwach w Serengeti")],
86 "current_agent": "researcher",
87 "task_complete": False
88})Checkpointing i Persistence
1from langgraph.checkpoint.memory import MemorySaver
2from langgraph.checkpoint.sqlite import SqliteSaver
3
4# Memory checkpoint (w pamięci)
5memory_checkpointer = MemorySaver()
6
7# SQLite checkpoint (trwały)
8sqlite_checkpointer = SqliteSaver.from_conn_string("safari_agent.db")
9
10# Kompilacja z checkpointerem
11app_with_memory = workflow.compile(checkpointer=memory_checkpointer)
12
13# Uruchomienie z thread_id dla kontynuacji
14config = {"configurable": {"thread_id": "safari-session-1"}}
15
16# Pierwsza wiadomość
17result1 = app_with_memory.invoke(
18 {"messages": [HumanMessage(content="Opowiedz mi o lwach")]},
19 config
20)
21
22# Kontynuacja rozmowy (ten sam thread_id)
23result2 = app_with_memory.invoke(
24 {"messages": [HumanMessage(content="A co z ich polowaniami?")]},
25 config
26)
27
28# Historia jest zachowana!
29print(result2["messages"])Human-in-the-loop
1from langgraph.graph import StateGraph, START, END
2from langgraph.checkpoint.memory import MemorySaver
3
4class HumanLoopState(TypedDict):
5 messages: Annotated[list, operator.add]
6 needs_approval: bool
7 approved: bool
8
9def agent_node(state: HumanLoopState) -> HumanLoopState:
10 """Agent proponuje akcję."""
11 response = llm.invoke(state["messages"])
12
13 # Sprawdź czy akcja wymaga zatwierdzenia
14 needs_approval = "delete" in response.content.lower() or "buy" in response.content.lower()
15
16 return {
17 "messages": [response],
18 "needs_approval": needs_approval
19 }
20
21def human_approval_node(state: HumanLoopState) -> HumanLoopState:
22 """Punkt zatwierdzenia przez człowieka."""
23 # Ten węzeł zatrzymuje wykonanie
24 # Użytkownik musi zatwierdzić przed kontynuacją
25 return state
26
27def execute_node(state: HumanLoopState) -> HumanLoopState:
28 """Wykonuje zatwierdzoną akcję."""
29 return {"messages": [AIMessage(content="Akcja wykonana!")]}
30
31def approval_router(state: HumanLoopState) -> Literal["human", "execute", "end"]:
32 """Router sprawdzający czy potrzeba zatwierdzenia."""
33 if state.get("needs_approval") and not state.get("approved"):
34 return "human"
35 elif state.get("approved"):
36 return "execute"
37 return "end"
38
39# Graf z human-in-the-loop
40human_loop_graph = StateGraph(HumanLoopState)
41
42human_loop_graph.add_node("agent", agent_node)
43human_loop_graph.add_node("human", human_approval_node)
44human_loop_graph.add_node("execute", execute_node)
45
46human_loop_graph.add_edge(START, "agent")
47human_loop_graph.add_conditional_edges("agent", approval_router)
48human_loop_graph.add_edge("human", "agent") # Po zatwierdzeniu wróć do agenta
49human_loop_graph.add_edge("execute", END)
50
51# Kompilacja z interrupt_before dla human node
52app = human_loop_graph.compile(
53 checkpointer=MemorySaver(),
54 interrupt_before=["human"] # Zatrzymaj przed human node
55)
56
57# Uruchomienie - zatrzyma się przed human node
58config = {"configurable": {"thread_id": "approval-1"}}
59result = app.invoke(
60 {"messages": [HumanMessage(content="Usuń wszystkie pliki")]},
61 config
62)
63
64# Sprawdź stan
65state = app.get_state(config)
66print(f"Needs approval: {state.values.get('needs_approval')}")
67
68# Użytkownik zatwierdza
69app.update_state(config, {"approved": True})
70
71# Kontynuuj wykonanie
72final_result = app.invoke(None, config)Branching i Parallel Execution
1from langgraph.graph import StateGraph, START, END
2from typing import Literal
3
4class ParallelState(TypedDict):
5 query: str
6 weather_result: str
7 animal_result: str
8 combined_result: str
9
10def weather_branch(state: ParallelState) -> ParallelState:
11 """Pobiera pogodę równolegle."""
12 # Symulacja wywołania API
13 return {"weather_result": "Pogoda: 28°C, słonecznie"}
14
15def animal_branch(state: ParallelState) -> ParallelState:
16 """Pobiera info o zwierzętach równolegle."""
17 return {"animal_result": "Zwierzęta: Lwy, słonie, żyrafy"}
18
19def combine_results(state: ParallelState) -> ParallelState:
20 """Łączy wyniki z równoległych gałęzi."""
21 combined = f"{state['weather_result']}. {state['animal_result']}"
22 return {"combined_result": combined}
23
24# Graf z równoległym wykonaniem
25parallel_graph = StateGraph(ParallelState)
26
27parallel_graph.add_node("weather", weather_branch)
28parallel_graph.add_node("animals", animal_branch)
29parallel_graph.add_node("combine", combine_results)
30
31# Równoległe krawędzie
32parallel_graph.add_edge(START, "weather")
33parallel_graph.add_edge(START, "animals")
34parallel_graph.add_edge("weather", "combine")
35parallel_graph.add_edge("animals", "combine")
36parallel_graph.add_edge("combine", END)
37
38parallel_app = parallel_graph.compile()
39
40result = parallel_app.invoke({"query": "Safari info"})
41print(result["combined_result"])Supervisor Pattern
1from langgraph.graph import StateGraph, START, END
2from typing import Literal
3
4class SupervisorState(TypedDict):
5 messages: Annotated[list, operator.add]
6 next_worker: str
7 completed_workers: list[str]
8
9def supervisor_node(state: SupervisorState) -> SupervisorState:
10 """Supervisor decyduje który worker ma działać."""
11 system_prompt = """Jesteś supervisorem zarządzającym zespołem:
12 - researcher: zbiera informacje
13 - analyst: analizuje dane
14 - writer: tworzy raporty
15
16 Na podstawie zadania i dotychczasowych wyników, wybierz następnego workera.
17 Odpowiedz TYLKO nazwą workera lub 'FINISH' jeśli zadanie ukończone."""
18
19 completed = state.get("completed_workers", [])
20
21 response = llm.invoke([
22 {"role": "system", "content": system_prompt},
23 {"role": "user", "content": f"Completed: {completed}\nTask: {state['messages'][-1].content}"}
24 ])
25
26 next_worker = response.content.strip().lower()
27
28 return {"next_worker": next_worker}
29
30def worker_node(worker_name: str):
31 """Fabryka węzłów workerów."""
32 def node(state: SupervisorState) -> SupervisorState:
33 prompts = {
34 "researcher": "Jesteś badaczem. Zbierz informacje.",
35 "analyst": "Jesteś analitykiem. Przeanalizuj dane.",
36 "writer": "Jesteś pisarzem. Napisz raport."
37 }
38
39 response = llm.invoke([
40 {"role": "system", "content": prompts[worker_name]},
41 *state["messages"]
42 ])
43
44 completed = state.get("completed_workers", []) + [worker_name]
45
46 return {
47 "messages": [AIMessage(content=f"[{worker_name}]: {response.content}")],
48 "completed_workers": completed
49 }
50 return node
51
52def supervisor_router(state: SupervisorState) -> Literal["researcher", "analyst", "writer", "end"]:
53 """Router supervisora."""
54 next_worker = state.get("next_worker", "").lower()
55 if next_worker == "finish" or next_worker == "end":
56 return "end"
57 return next_worker
58
59# Budowa grafu supervisora
60supervisor_graph = StateGraph(SupervisorState)
61
62supervisor_graph.add_node("supervisor", supervisor_node)
63supervisor_graph.add_node("researcher", worker_node("researcher"))
64supervisor_graph.add_node("analyst", worker_node("analyst"))
65supervisor_graph.add_node("writer", worker_node("writer"))
66
67supervisor_graph.add_edge(START, "supervisor")
68
69# Wszystkie krawędzie prowadzą z powrotem do supervisora
70for worker in ["researcher", "analyst", "writer"]:
71 supervisor_graph.add_edge(worker, "supervisor")
72
73supervisor_graph.add_conditional_edges(
74 "supervisor",
75 supervisor_router,
76 {
77 "researcher": "researcher",
78 "analyst": "analyst",
79 "writer": "writer",
80 "end": END
81 }
82)
83
84supervisor_app = supervisor_graph.compile()
85
86# Uruchomienie
87result = supervisor_app.invoke({
88 "messages": [HumanMessage(content="Przygotuj kompletny raport o Safari w Serengeti")],
89 "completed_workers": []
90})LangGraph to potężne narzędzie do budowania złożonych przepływów AI. W następnej lekcji poznasz subagentów i hierarchie agentów!