The right way to Construct a Conversational Analysis AI Agent with LangGraph: Step Replay and Time-Journey Checkpoints

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


import os
import json
import textwrap
import getpass
import time
from typing import Annotated, Listing, Dict, Any, Optionally available


from typing_extensions import TypedDict


from langchain.chat_models import init_chat_model
from langchain_core.messages import BaseMessage
from langchain_core.instruments import device


from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.reminiscence import InMemorySaver
from langgraph.prebuilt import ToolNode, tools_condition


import requests
from requests.adapters import HTTPAdapter, Retry


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


llm = init_chat_model("google_genai:gemini-2.0-flash")

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


_session = requests.Session()
_session.headers.replace({
   "Consumer-Agent": "LangGraph-Colab-Demo/1.0 (contact: [email protected])",
   "Settle for": "utility/json",
})
retry = Retry(
   complete=5, join=5, learn=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(question: str, restrict: int = 3) -> Listing[Dict[str, str]]:
   """
   Use MediaWiki search API with:
     - origin='*' (good follow for CORS)
     - Well mannered UA + retries
   Returns compact record of {title, snippet_html, url}.
   """
   params = {
       "motion": "question",
       "record": "search",
       "format": "json",
       "srsearch": question,
       "srlimit": restrict,
       "srprop": "snippet",
       "utf8": 1,
       "origin": "*",
   }
   r = _session.get(WIKI_SEARCH_URL, params=params, timeout=15)
   r.raise_for_status()
   information = r.json()
   out = []
   for merchandise in information.get("question", {}).get("search", []):
       title = merchandise.get("title", "")
       page_url = f"https://en.wikipedia.org/wiki/{title.change(' ', '_')}"
       snippet = merchandise.get("snippet", "")
       out.append({"title": title, "snippet_html": snippet, "url": page_url})
   return out


@device
def wiki_search(question: str) -> Listing[Dict[str, str]]:
   """Search Wikipedia and return as much as 3 outcomes with title, snippet_html, and url."""
   strive:
       outcomes = _wiki_search_raw(question, restrict=3)
       return outcomes if outcomes else [{"title": "No results", "snippet_html": "", "url": ""}]
   besides Exception as e:
       return [{"title": "Error", "snippet_html": str(e), "url": ""}]


TOOLS = [wiki_search]

class State(TypedDict):
   messages: Annotated[list, add_messages]


graph_builder = StateGraph(State)


llm_with_tools = llm.bind_tools(TOOLS)


SYSTEM_INSTRUCTIONS = textwrap.dedent("""
You're ResearchBuddy, a cautious analysis assistant.
- If the person asks you to "analysis", "discover data", "newest", "internet", or references a library/framework/product,
 you SHOULD name the `wiki_search` device at the least as soon as earlier than finalizing your reply.
- Once you name instruments, be concise within the textual content you produce across the name.
- After receiving device outcomes, cite at the least the web page titles you utilized in your abstract.
""").strip()


def chatbot(state: State) -> Dict[str, Any]:
   """Single step: name the LLM (with instruments certain) on the present messages."""
   return {"messages": [llm_with_tools.invoke(state["msgs"])]}


graph_builder.add_node("chatbot", chatbot)


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

def print_last_message(occasion: Dict[str, Any]):
   """Fairly-print the final message in an occasion if out there."""
   if "messages" in occasion and occasion["messages"]:
       msg = occasion["messages"][-1]
       strive:
           if isinstance(msg, BaseMessage):
               msg.pretty_print()
           else:
               position = msg.get("position", "unknown")
               content material = msg.get("content material", "")
               print(f"n[{role.upper()}]n{content material}n")
       besides Exception:
           print(str(msg))


def show_state_history(cfg: Dict[str, Any]) -> Listing[Any]:
   """Print a concise view of checkpoints; return the record as nicely."""
   historical past = record(graph.get_state_history(cfg))
   print("n=== 📜 State historical past (most up-to-date first) ===")
   for i, st in enumerate(historical past):
       n = st.subsequent
       n_txt = f"{n}" if n else "()"
       print(f"{i:02d}) NumMessages={len(st.values.get('messages', []))}  Subsequent={n_txt}")
   print("=== Finish historical past ===n")
   return historical past


def pick_checkpoint_by_next(historical past: Listing[Any], node_name: str = "instruments") -> Optionally available[Any]:
   """Decide the primary checkpoint whose `subsequent` features a given node (e.g., 'instruments')."""
   for st in historical past:
       nxt = tuple(st.subsequent) if st.subsequent else tuple()
       if node_name in nxt:
           return 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 person flip ====================")
occasions = graph.stream(first_turn, config, stream_mode="values")
for ev in occasions:
   print_last_message(ev)


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


print("n==================== 🟢 STEP 2: Second person flip ====================")
occasions = graph.stream(second_turn, config, stream_mode="values")
for ev in occasions:
   print_last_message(ev)

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


to_replay = pick_checkpoint_by_next(historical past, node_name="instruments")
if to_replay is None:
   to_replay = historical past[min(2, len(history) - 1)]


print("Chosen checkpoint to renew from:")
print("  Subsequent:", to_replay.subsequent)
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 just isn't None and 0 <= MANUAL_INDEX < len(historical past):
   chosen = historical past[MANUAL_INDEX]
   print(f"n==================== 🧭 MANUAL RESUME @ index {MANUAL_INDEX} ====================")
   print("Subsequent:", chosen.subsequent)
   print("Config:", chosen.config)
   for ev in graph.stream(None, chosen.config, stream_mode="values"):
       print_last_message(ev)


print("n✅ Completed. You added steps, replayed historical past, and resumed from a previous checkpoint.")