How to Build an AI Agent with LangGraph

1"""LangGraph research agent: plan -> search -> synthesize -> human review loop -> finalize.
2Run: python research_agent_graph.py (pip install langgraph anthropic; set ANTHROPIC_API_KEY)
3"""
4import uuid
5from typing import Optional
6
7import anthropic
8from langgraph.checkpoint.memory import InMemorySaver
9from langgraph.graph import END, START, StateGraph
10from langgraph.types import Command, interrupt
11from typing_extensions import TypedDict
12
13client = anthropic.Anthropic()
14
15
16class State(TypedDict):
17    topic: str
18    plan: str
19    search_results: str
20    draft: str
21    decision: Optional[str]
22    feedback: Optional[str]
23
24
25def plan_node(state: State) -> dict:
26    resp = client.messages.create(
27        model="claude-sonnet-5-5",
28        max_tokens=300,
29        messages=[{"role": "user", "content": f"List 3 bullet points to research for a short report on: {state['topic']}"}],
30    )
31    return {"plan": resp.content[0].text}
32
33
34def search_node(state: State) -> dict:
35    # Stub: swap for a real search API call (Tavily, Exa, Brave Search) in production
36    return {"search_results": f"Findings for '{state['topic']}' based on plan:\n{state['plan']}"}
37
38
39def synthesize_node(state: State) -> dict:
40    feedback_note = f"\nIncorporate this feedback: {state['feedback']}" if state.get("feedback") else ""
41    resp = client.messages.create(
42        model="claude-sonnet-5-5",
43        max_tokens=600,
44        messages=[{"role": "user", "content": f"Write a short report from these findings.{feedback_note}\n\n{state['search_results']}"}],
45    )
46    return {"draft": resp.content[0].text}
47
48
49def review_node(state: State) -> dict:
50    decision = interrupt({"draft": state["draft"], "ask": "approve or request_changes?"})
51    if decision.get("action") == "approve":
52        return {"decision": "approve"}
53    return {"decision": "request_changes", "feedback": decision.get("feedback", "")}
54
55
56def finalize_node(state: State) -> dict:
57    return {"draft": state["draft"] + "\n\n[Approved and finalized]"}
58
59
60def route_after_review(state: State) -> str:
61    return "finalize" if state["decision"] == "approve" else "synthesize"
62
63
64builder = StateGraph(State)
65builder.add_node("plan", plan_node)
66builder.add_node("search", search_node)
67builder.add_node("synthesize", synthesize_node)
68builder.add_node("review", review_node)
69builder.add_node("finalize", finalize_node)
70builder.add_edge(START, "plan")
71builder.add_edge("plan", "search")
72builder.add_edge("search", "synthesize")
73builder.add_edge("synthesize", "review")
74builder.add_conditional_edges("review", route_after_review, {"finalize": "finalize", "synthesize": "synthesize"})
75builder.add_edge("finalize", END)
76
77graph = builder.compile(checkpointer=InMemorySaver())
78
79
80if __name__ == "__main__":
81    config = {"configurable": {"thread_id": str(uuid.uuid4())}}
82    result = graph.invoke({"topic": "edge AI chips", "feedback": None}, config)
83
84    if "__interrupt__" in result:
85        print("Draft for review:\n", result["__interrupt__"][0].value["draft"])
86        result = graph.invoke(
87            Command(resume={"action": "request_changes", "feedback": "Add a cost comparison section."}),
88            config,
89        )
90
91    if "__interrupt__" in result:
92        print("\nNew draft for review:\n", result["__interrupt__"][0].value["draft"])
93        result = graph.invoke(Command(resume={"action": "approve"}), config)
94
95    print("\nFinal:\n", result["draft"])
96