Learn how to build a conversational AI agent with LangGraph using step replay and time-travel checkpoints

!pip -q install -U langgraph langchain langchain-google-genai google-generativeai typing_extensions
!pip -q install "requests==2.32.4"


Import os
Import json
Textwrap
import getpass
import time
Type import to get Annotated List, Dict or Optional, and any other type of list.


Import TypedDict from typing_extensions


from langchain.chat_models import init_chat_model
from langchain_core.messages import BaseMessage
Import tool from langchain_core.tools


StateGraph import from langgraph.graph, START and END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver
From langgraph.prebuilt, import Tools_condition


Requests for imports
Retry HTTPAdapter can be imported by importing requests.adapters.


If not, os.environ.get ("GOOGLE_API_KEY"):
   os.environ["GOOGLE_API_KEY"] = getpass.getpass("🔑 Enter your Google API Key (Gemini): ")


The llm model is initialized by llm."google_genai:gemini-2.0-flash")

WIKI_SEARCH_URL = "https://en.wikipedia.org/w/api.php"


_session = requests.Session()
_session.headers.update({
   "User-Agent": "LangGraph-Colab-Demo/1.0 (contact: [email protected])",
   "Accept": "application/json",
})
Retry = (
   total=5, connect=5, read=5, backoff_factor=0.5,
   status_forcelist=(429, 500, 502, 503, 504),
   allowed_methods=("GET", "POST")
)
_session.mount("https://", HTTPAdapter(max_retries=retry))
_session.mount("http://", HTTPAdapter(max_retries=retry))


def _wiki_search_raw(query: str, limit: int = 3) -> List[Dict[str, str]]:
   """
 You can use the MediaWiki Search API to:
     - origin='*' (good practice for CORS)
 The retry option is available for both the Polite UA and the retries."params" =
   Returns compact list of {title, snippet_html, url}.
   """
   params = {
       "action": "query",
       "list": "search",
       "format": "json",
       "srsearch": query,
       "srlimit": limit,
       "srprop": "snippet",
       "utf8": 1,
       "origin": "*",
   }
   r = _session.get(WIKI_SEARCH_URL, params=params, timeout=15)
   r.raise_for_status()
 Data = r.json()
 You can also say: []
 Get item from data"query", {}).get("search", []):
 The title is item.get"title", "")
 Page_url = "f""https://en.wikipedia.org/wiki/{title.replace(' ', '_')}"
 The snippet is item.get("snippet", "")
       out.append({"title": title, "snippet_html": snippet, "url": page_url})
 Return out


@tool
def wiki_search(query: str) -> List[Dict[str, str]]:
   """Search Wikipedia and return up to 3 results with title, snippet_html, and url."""
   try:
       results = _wiki_search_raw(query, limit=3)
       return results if results else [{"title": "No results", "snippet_html": "", "url": ""}]
 Except Exception As e.
 You can return to your original language by clicking here. [{"title": "Error", "snippet_html": str(e), "url": ""}]


TOOLS [wiki_search]

"class State" (TypedDict, Typed Dictionary)
 All messages are Annotated[list, add_messages]


graph_builder = StateGraph(State)


llm_with_tools = llm.bind_tools(TOOLS)


SYSTEM_INSTRUCTIONS = textwrap.dedent("""
ResearchBuddy is your diligent assistant.
If you are asked to do so by the user "research", "find info", "latest", "web", or references a library/framework/product,
 Before settling on a final answer, you should call the wiki_search search tool.
Be concise when you create the text around your call.
After you receive the tool results, at least cite page titles that were used to create your summary.Return
""").strip()


def chatbot(state: State) -> Dict[str, Any]:
   """Single step: call the LLM (with tools bound) on the current messages."""
   return {"messages": [llm_with_tools.invoke(state["msgs"])]}


graph_builder.add_node("chatbot", chatbot)


InMemorySaver = memory()
graph = graph_builder.compile(checkpointer=memory)

def print_last_message(event: Dict[str, Any]):
   """Pretty-print the last message in an event if available."""
 If you want to know more about if "messages" The event is a big deal["messages"]:
 Event = msg["messages"][-1]
       try:
 If isInstance(msg BaseMessage), then:
               msg.pretty_print()
           else:
 A role is msg.get()"role", "unknown")
 Get the content of msg.get()"content", "")
               print(f"n[{role.upper()}]n{content}n")
 The exception:
           print(str(msg))


def show_state_history(cfg: Dict[str, Any]) -> List[Any]:
   """Print a concise view of checkpoints; return the list as well."""
   history = list(graph.get_state_history(cfg))
   print("n=== 📜 State history (most recent first) ===")
 The enumerate (history) is for the i and st:
 Next = n
 N_txt=f"{n}" if n else "()"
       print(f"{i:02d}) NumMessages={len(st.values.get('messages', []))}  Next={n_txt}")
   print("=== End history ===n")
 Return History


def pick_checkpoint_by_next(history: List[Any]Node_name : str "tools") -> Optional[Any]:
   """Pick the first checkpoint whose `next` includes a given node (e.g., 'tools')."""
 For st, in history
 If st.next then tuple, else nxt.()
 Node_name in the nxt
 The return of the st
 Return None

config = {"configurable": {"thread_id": "demo-thread-1"}}


first_turn = {
   "messages": [
       {"role": "system", "content": SYSTEM_INSTRUCTIONS},
       {"role": "user", "content": "I'm learning LangGraph. Could you do some research on it for me?"},
   ]
}


print("n==================== 🟢 STEP 1: First user turn ====================")
events = graph.stream(first_turn, config, stream_mode="values")
For ev, see events
   print_last_message(ev)


second_turn = {
   "messages": [
       {"role": "user", "content": "Ya. Maybe I'll build an agent with it!"}
   ]
}


print("n==================== 🟢 STEP 2: Second user turn ====================")
events = graph.stream(second_turn, config, stream_mode="values")
For ev, see events
   print_last_message(ev)

print("n==================== 🔁 REPLAY: Full state history ====================")
history = show_state_history(config)


to_replay = pick_checkpoint_by_next(history, node_name="tools")
If to_replay = None
 To_replay is history[min(2, len(history) - 1)]


print("Chosen checkpoint to resume from:")
print("  Next:", to_replay.next)
print("  Config:", to_replay.config)


print("n==================== ⏪ RESUME from chosen checkpoint ====================")
For ev in graph.stream. (None. to_replay.config. stream_mode="vals"):
   print_last_message(ev)


Manual_Index = None 
If MANUAL_INDEX ain't None, then it will be 0.