How to create social media AI agent in Langgraph: Retention action and test zones
In this lesson, we intend to understand how Langgraming helps us to manage a discussion in an organized way, while and we empower “in the past” through testing areas. By building Chatbot that includes free Gemini model and Wikipedia tool, we can add more steps to the discussion, recording each other, and again starting with full history. This approach is enabling us to see, in real time, the way Langbram design has followed and deception for discussion and control. Look Full codes here.
!pip -q install -U langgraph langchain langchain-google-genai google-generativeai typing_extensions
!pip -q install "requests==2.32.4"
import os
import json
import textwrap
import getpass
import time
from typing import Annotated, List, Dict, Any, Optional
from typing_extensions import TypedDict
from langchain.chat_models import init_chat_model
from langchain_core.messages import BaseMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory 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")We start by installing the required libraries, setting our Gemini API key, and importing all required modules. We are starting the Gemini model using Langchain to use it as a LLM ILM in our Langgraph operations. Look Full codes here.
WIKI_SEARCH_URL = "
_session = requests.Session()
_session.headers.update({
"User-Agent": "LangGraph-Colab-Demo/1.0 (contact: [email protected])",
"Accept": "application/json",
})
retry = Retry(
total=5, connect=5, read=5, backoff_factor=0.5,
status_forcelist=(429, 500, 502, 503, 504),
allowed_methods=("GET", "POST")
)
_session.mount(" HTTPAdapter(max_retries=retry))
_session.mount(" HTTPAdapter(max_retries=retry))
def _wiki_search_raw(query: str, limit: int = 3) -> List[Dict[str, str]]:
"""
Use MediaWiki search API with:
- origin='*' (good practice for CORS)
- Polite UA + retries
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()
out = []
for item in data.get("query", {}).get("search", []):
title = item.get("title", "")
page_url = f" ', '_')}"
snippet = 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:
return [{"title": "Error", "snippet_html": str(e), "url": ""}]
TOOLS = [wiki_search]We put the Wikipedia search tool with custom session, restoration, and good user agent. We explain _Search_search_raw to ask the mediawiki API and threaten the Langchain tool, it gives us a seamless tool to call the seams into the functioning of our Langgraph operating work. Look Full codes here.
class State(TypedDict):
messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)
llm_with_tools = llm.bind_tools(TOOLS)
SYSTEM_INSTRUCTIONS = textwrap.dedent("""
You are ResearchBuddy, a careful research assistant.
- If the user asks you to "research", "find info", "latest", "web", or references a library/framework/product,
you SHOULD call the `wiki_search` tool at least once before finalizing your answer.
- When you call tools, be concise in the text you produce around the call.
- After receiving tool results, cite at least the page titles you used in your summary.
""").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)
memory = InMemorySaver()
graph = graph_builder.compile(checkpointer=memory)It describes our graph condition to keep the rope of a running message and bind our Gemini model to Wiki_search tools, allowing you to call when needed. We include Chatbot and no and node tools, add them to two edges, and enable the final look at the In-Memory Saver. We now combine graphs to add steps, repeat history, and resume from any testing site. Look Full codes here.
def print_last_message(event: Dict[str, Any]):
"""Pretty-print the last message in an event if available."""
if "messages" in event and event["messages"]:
msg = event["messages"][-1]
try:
if isinstance(msg, BaseMessage):
msg.pretty_print()
else:
role = msg.get("role", "unknown")
content = msg.get("content", "")
print(f"n[{role.upper()}]n{content}n")
except 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) ===")
for i, st in enumerate(history):
n = st.next
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:
nxt = tuple(st.next) if st.next else tuple()
if node_name in nxt:
return st
return NoneWe include use functions to make our Langgraph's function easier to look and control. We use printing_last_message to show the most recent answer, show_Sistory to list all checkpoints, and select_de_Next finding a test site, such as action tools. Look Full codes here.
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 in 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 in events:
print_last_message(ev)We imitate two user interactions in the same medicine by distribution of events with graphs. First we provide program commands and ask the assistant to research Langstraph, and follow the second user message by creating independent agent. Each step is there, allowing us to use or resume the provinces later. Look Full codes here.
print("n==================== 🔁 REPLAY: Full state history ====================")
history = show_state_history(config)
to_replay = pick_checkpoint_by_next(history, node_name="tools")
if to_replay is None:
to_replay = 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 is not None and 0 <= MANUAL_INDEX < len(history):
chosen = history[MANUAL_INDEX]
print(f"n==================== 🧭 MANUAL RESUME @ index {MANUAL_INDEX} ====================")
print("Next:", chosen.next)
print("Config:", chosen.config)
for ev in graph.stream(None, chosen.config, stream_mode="values"):
print_last_message(ev)
print("n✅ Done. You added steps, replayed history, and resumed from a prior checkpoint.")We restore the full checkup history to see how our conversation comes from steps and has identified a useful point to restart. Then “We are going for time” by resetting from the selected area, and optional from any manual indicator, so we continue the discussion directly from the saved scenario.
In conclusion, we found a clear picture that the LangSgraph Assessment skills and travel skills brings the variable and visibility of negotiations. By treading about a lot of user exchange, an overload of the State History, and restarts previous Points, we can see firsthand the framework for building researchers or private assistants. We realize that this work movement is not just short, but the foundation we can extend to the most complex applications, where production and registration is important as the answers.
Look Full codes here. Feel free to look our GITHUB page for tutorials, codes and letters of writing. Also, feel free to follow it Sane and don't forget to join ours 100K + ml subreddit Then sign up for Our newspaper.
Asphazzaq is a Markteach Media Inc. According to a View Business and Developer, Asifi is committed to integrating a good social intelligence. His latest attempt is launched by the launch of the chemistrylife plan for an intelligence, MarktechPost, a devastating intimate practice of a machine learning and deep learning issues that are clearly and easily understood. The platform is adhering to more than two million moon visits, indicating its popularity between the audience.
