A Groq-Powered Agentic Research Assistant with LangGraph, Tool Calling, Sub-Agents and Agentic Memory

This tutorial builds a Groq-powered agentic research workflow that runs against Groq’s free, OpenAI-compatible inference endpoint. (The backend here is Groq, the high-speed inference provider, rather than xAI’s similarly named Grok model.) LangChain’s ChatOpenAI interface is configured to work with Groq by setting the Groq API key and base URL, which makes it possible to use fast hosted open models such as llama-3.3-70b-versatile for tool-based reasoning. The model is then connected to practical tools for web search, webpage fetching, file handling, Python execution, skill loading, sub-agent delegation and long-term memory. The end result is a working multi-step agent that can research a topic, assign focused sub-tasks, generate structured output and save useful information for later runs.

import subprocess, sys
def _pip(*a): subprocess.check_call((sys.executable,"-m","pip","install","-q",*a))
_pip("langgraph>=0.2.50", "langchain>=0.3.0", "langchain-openai>=0.2.0",
    "langchain-community>=0.3.0", "ddgs", "requests", "beautifulsoup4",
    "tiktoken", "pydantic>=2.0")


import os, getpass
if not os.environ.get("GROQ_API_KEY"):
   os.environ("GROQ_API_KEY") = getpass.getpass("GROQ_API_KEY (free at console.groq.com/keys): ")


os.environ("OPENAI_API_KEY")  = os.environ("GROQ_API_KEY")
os.environ("OPENAI_BASE_URL") = "https://api.groq.com/openai/v1"


MODEL_NAME = "llama-3.3-70b-versatile"


import json, re, io, contextlib, pathlib
from typing import Annotated, TypedDict, Sequence, Literal, List, Dict, Any
from datetime import datetime, timezone
from langchain_openai import ChatOpenAI
from langchain_core.messages import (
   SystemMessage, HumanMessage, AIMessage, ToolMessage, BaseMessage)
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode

The first step installs the core libraries needed to build the workflow, including LangGraph, LangChain, DuckDuckGo search utilities and supporting parsing libraries, then securely collects the Groq API key and configures Groq as an OpenAI-compatible endpoint by setting the appropriate key and base URL. The necessary modules for messaging, tools, graph creation, typing, file-system handling and model initialisation are imported alongside it.

SANDBOX = pathlib.Path("/content/deerflow_sandbox").resolve()
for sub in ("uploads","workspace","outputs","skills/public","skills/custom","memory"):
   (SANDBOX/sub).mkdir(parents=True, exist_ok=True)


def _safe(p: str) -> pathlib.Path:
   full = (SANDBOX/p.lstrip("/")).resolve()
   if not str(full).startswith(str(SANDBOX)):
       raise ValueError(f"path escapes sandbox: {p}")
   return full


SKILLS: Dict(str, Dict(str,str)) = {}
def register_skill(name, description, content, location="public"):
   d = SANDBOX/"skills"/location/name; d.mkdir(parents=True, exist_ok=True)
   (d/"SKILL.md").write_text(content)
   SKILLS(name) = {"description": description, "content": content,
                   "path": str(d/"SKILL.md")}


register_skill("research",
   "Conduct multi-source web research on a topic and produce structured notes.",
   """# Research Skill
## Workflow
1. Decompose the question into 3-5 sub-questions.
2. For each sub-question call `web_search` and pick 2 authoritative URLs.
3. `web_fetch` those URLs; extract concrete facts, numbers, dates.
4. Cross-reference for consensus vs. disagreement.
5. Append findings to `workspace/research_notes.md`: claim → evidence → URL.
## Best practices
- Prefer primary sources. Note dates. Never fabricate URLs or numbers.""")


register_skill("report-generation",
   "Synthesize research notes into a polished markdown report in outputs/.",
   """# Report Generation Skill
## Workflow
1. file_read('workspace/research_notes.md').
2. Outline: exec summary, key findings, analysis, conclusion, sources.
3. file_write('outputs/report.md', ...).
## Structure
- # Title
- ## Executive Summary  (3–5 sentences)
- ## Key Findings       (bullets)
- ## Detailed Analysis  (sections)
- ## Conclusion
- ## Sources            (numbered URL list)""")


register_skill("code-execution",
   "Run Python in the sandbox for computation, data wrangling, charts.",
   """# Code Execution Skill
1. Plan in plain language first.
2. python_exec the code; persistent artifacts go to /outputs/.
3. Verify before quoting results.""")


MEM = SANDBOX/"memory/long_term.json"
if not MEM.exists():
   MEM.write_text(json.dumps({"facts":(),"preferences":{}}, indent=2))
def _load_mem(): return json.loads(MEM.read_text())
def _save_mem(m): MEM.write_text(json.dumps(m, indent=2))

A sandboxed project directory is created — in this example within Google Colab — to keep uploads, workspace files, outputs, skills and memory organised in one controlled location. Reusable skills for research, report generation and code execution are defined so that agents can discover and follow structured workflows, and a simple long-term memory JSON file is initialised to store facts and preferences across multiple runs within the same sandbox.

@tool
def list_skills() -> str:
   """List all skills with one-line descriptions. Call this first for complex tasks."""
   return "n".join(f"- {n}: {s('description')}" for n,s in SKILLS.items())


@tool
def load_skill(name: str) -> str:
   """Load full SKILL.md for `name`. Call before running its workflow."""
   if name not in SKILLS: return f"Unknown. Available: {list(SKILLS)}"
   return SKILLS(name)("content")


@tool
def web_search(query: str, max_results: int = 5) -> str:
   """Search the web (DuckDuckGo). Returns titles, URLs, snippets."""
   from ddgs import DDGS
   out = ()
   try:
       with DDGS() as d:
           for r in d.text(query, max_results=max_results):
               out.append(f"- {r.get('title','')}n  URL: {r.get('href','')}n  "
                          f"{(r.get('body') or '')(:220)}")
   except Exception as e:
       return f"search error: {e}"
   return "n".join(out) or "no results"


@tool
def web_fetch(url: str, max_chars: int = 4000) -> str:
   """Fetch a URL, return cleaned text (scripts/nav stripped)."""
   import requests
   from bs4 import BeautifulSoup
   try:
       r = requests.get(url, timeout=15,
                        headers={"User-Agent":"Mozilla/5.0 DeerFlow-Lite"})
       soup = BeautifulSoup(r.text, "html.parser")
       for s in soup(("script","style","nav","footer","aside","header")): s.decompose()
       text = re.sub(r"ns*n", "nn", soup.get_text("n")).strip()
       return text(:max_chars) or "(empty page)"
   except Exception as e:
       return f"fetch error: {e}"


@tool
def file_write(path: str, content: str) -> str:
   """Write content to a sandbox path, e.g. 'workspace/notes.md' or 'outputs/x.md'."""
   p = _safe(path); p.parent.mkdir(parents=True, exist_ok=True)
   p.write_text(content)
   return f"wrote {len(content)} chars → {path}"


@tool
def file_read(path: str) -> str:
   """Read a sandbox file (first 8 KB)."""
   p = _safe(path)
   return p.read_text()(:8000) if p.exists() else f"not found: {path}"


@tool
def file_list(path: str = "") -> str:
   """List files under a sandbox dir."""
   base = _safe(path) if path else SANDBOX
   if not base.exists(): return "not found"
   items = ()
   for c in sorted(base.rglob("*")):
       if "memory" in c.relative_to(SANDBOX).parts: continue
       items.append(f"  {'D' if c.is_dir() else 'F'}  {c.relative_to(SANDBOX)}")
   return "n".join(items(:60)) or "(empty)"


@tool
def python_exec(code: str) -> str:
   """Run Python in the sandbox. SANDBOX_ROOT is preset."""
   g = {"__name__":"__sb__", "SANDBOX_ROOT": str(SANDBOX)}
   buf = io.StringIO()
   try:
       with contextlib.redirect_stdout(buf), contextlib.redirect_stderr(buf):
           exec(code, g)
       return (buf.getvalue() or "(no stdout)")(:4000)
   except Exception as e:
       return f"{type(e).__name__}: {e}n{buf.getvalue()(:1500)}"


@tool
def remember(fact: str) -> str:
   """Persist a single fact to long-term memory (survives across runs)."""
   m = _load_mem()
   m("facts").append({"fact": fact, "ts": datetime.now(timezone.utc).isoformat()})
   _save_mem(m)
   return f"remembered ({len(m('facts'))} total)"


@tool
def recall() -> str:
   """Retrieve everything in long-term memory."""
   m = _load_mem()
   if not m("facts"): return "(memory empty)"
   return "n".join(f"- {f('fact')}" for f in m("facts")(-20:))

The main tools that the agent can call during execution are then defined: listing skills, loading skill instructions, searching the web, fetching webpages, reading files and writing files. The agent is also given a sandboxed Python execution environment so it can run calculations or generate artifacts as needed, along with memory tools that let it record important facts and recall previously stored information.

@tool
def spawn_subagent(role: str, task: str,
                  allowed_tools: str = "web_search,web_fetch,file_write,file_read") -> str:
   """Spawn an isolated sub-agent with a focused role and scoped tools.
   Returns its final report string. Use for parallelizable / focused subtasks."""
   bag = {t.name: t for t in BASE_TOOLS}
   sub_tools = (bag(n.strip()) for n in allowed_tools.split(",") if n.strip() in bag)
   sub_llm = ChatOpenAI(model=MODEL_NAME, temperature=0.2).bind_tools(sub_tools)
   sys_msg = SystemMessage(content=(
       f"You are a specialized sub-agent. Role: {role}.n"
       f"You operate in an ISOLATED context — no access to lead history.n"
       f"Tools: {', '.join(t.name for t in sub_tools)}.n"
       "End with a final assistant message starting 'FINAL REPORT:' "
       "containing a structured ≤700-word summary including any URLs."))
   msgs: List(BaseMessage) = (sys_msg, HumanMessage(content=task))
   for _ in range(8):
       r = sub_llm.invoke(msgs); msgs.append(r)
       if not getattr(r, "tool_calls", None):
           return f"(sub-agent: {role})n" + (r.content if isinstance(r.content,str) else str(r.content))
       for tc in r.tool_calls:
           t = bag.get(tc("name"))
           try:
               res = t.invoke(tc("args")) if t else f"unknown tool {tc('name')}"
           except Exception as e:
               res = f"tool error: {e}"
           msgs.append(ToolMessage(content=str(res)(:3000), tool_call_id=tc("id")))
   return f"(sub-agent: {role}) step-limit reached."


BASE_TOOLS = (list_skills, load_skill, web_search, web_fetch, file_write,
             file_read, file_list, python_exec, remember, recall)
ALL_TOOLS = BASE_TOOLS + (spawn_subagent)


LEAD_SYSTEM = f"""You are DeerFlow-Lite, a long-horizon super-agent harness.


Sandbox layout (relative to {SANDBOX}):
 uploads/    – user files
 workspace/  – your scratchpad
 outputs/    – final deliverables
 skills/     – capability modules (load_skill)


Principles:
 • For non-trivial tasks: list_skills → load_skill → execute.
 • Use spawn_subagent for focused subtasks (isolated context keeps lead lean).
 • Persist intermediates to workspace/, deliverables to outputs/.
 • Use remember(fact) for cross-session knowledge.
 • Finish with a short summary of what was produced and where.


Today: {datetime.now(timezone.utc).strftime('%Y-%m-%d')}."""


class AgentState(TypedDict):
   messages: Annotated(Sequence(BaseMessage), add_messages)


llm = ChatOpenAI(model=MODEL_NAME, temperature=0.3).bind_tools(ALL_TOOLS)


def call_model(state: AgentState):
   msgs = list(state("messages"))
   if not msgs or not isinstance(msgs(0), SystemMessage):
       msgs = (SystemMessage(content=LEAD_SYSTEM)) + msgs
   return {"messages": (llm.invoke(msgs))}


def route(state: AgentState) -> Literal("tools","__end__"):
   last = state("messages")(-1)
   return "tools" if getattr(last, "tool_calls", None) else END


g = StateGraph(AgentState)
g.add_node("agent", call_model)
g.add_node("tools", ToolNode(ALL_TOOLS))
g.set_entry_point("agent")
g.add_conditional_edges("agent", route, {"tools":"tools", END: END})
g.add_edge("tools", "agent")
APP = g.compile()

Next, the available tools are collected, a lead system prompt is defined, the Groq-backed chat model is initialised, and the tools are bound to it. The LangGraph workflow is assembled so the agent can alternate between reasoning and tool execution until it reaches a final answer.

def run(task: str, max_steps: int = 25):
   print("="*78); print(f"🦌 TASK: {task}"); print("="*78)
   state = {"messages":(HumanMessage(content=task))}
   n = 0
   for ev in APP.stream(state, {"recursion_limit": max_steps*2}, stream_mode="updates"):
       for node, payload in ev.items():
           for m in payload.get("messages", ()):
               n += 1
               if isinstance(m, AIMessage):
                   if m.tool_calls:
                       for tc in m.tool_calls:
                           args = json.dumps(tc("args"), ensure_ascii=False)
                           args = args(:140) + ("…" if len(args)>140 else "")
                           print(f"({n:02}) 🔧 {tc('name')}({args})")
                   else:
                       txt = m.content if isinstance(m.content,str) else str(m.content)
                       print(f"({n:02}) 🦌 {txt(:800)}")
               elif isinstance(m, ToolMessage):
                   s = str(m.content).replace("n"," ")(:220)
                   print(f"({n:02}) 📤 {s}")
   print("n"+"="*78); print("✅ COMPLETE — sandbox state:"); print("="*78)
   print(file_list.invoke({"path":""}))
   print("n🧠 Long-term memory:"); print(recall.invoke({}))
   for f in sorted((SANDBOX/"outputs").rglob("*")):
       if f.is_file():
           print(f"n--- 📄 {f.relative_to(SANDBOX)} (first 800 chars) ---")
           print(f.read_text()(:800))


run(
   "Give me a briefing on small language models (SLMs) in 2025. "
   "(1) discover skills; (2) spawn a researcher sub-agent to gather "
   "specifics on three notable SLMs from 2024-2025 with sizes, benchmarks, "
   "and use cases — sub-agent saves to workspace/slm_research.md; "
   "(3) load report-generation skill and write outputs/slm_briefing.md "
   "(~400 words) with a Sources section; (4) save the single most "
   "important takeaway to long-term memory; (5) summarize.",
   max_steps=25,
)

Finally, a run() function starts a user task, streams each agent step, and prints tool calls, tool outputs and final responses in a readable format, then displays the sandbox file structure, long-term memory and generated output files once the workflow completes. A demonstration task has the agent research small language models, prepare a briefing, save a report and store a key takeaway in memory.

Taken together, the example is a compact but capable agent framework showing how Groq’s OpenAI-compatible API can serve as a fast, accessible backend for advanced LLM workflows, with LangGraph managing the agent loop and LangChain binding the tools.

Limitations and what to watch

This is an educational reference rather than a production system. It relies on Groq’s free tier, which is rate-limited — the llama-3.3-70b-versatile model, for example, is capped at a set number of requests per day on the free plan — so heavier or commercial use would require a paid tier and careful rate-limit handling. The sandbox runs inside a notebook environment such as Colab, where files and memory do not persist once the session ends, and the agent’s web search and Python execution tools should be treated with the usual caution any time a model can fetch external content or run code.

Agentic reliability is also an open question: multi-step agents can loop, call tools incorrectly or produce confidently wrong output, so human review of generated reports remains necessary, and model names, endpoints and free-tier limits change over time and should be confirmed against current Groq documentation before building on them. Background on the model and endpoint is available from Groq’s documentation. Readers interested in giving such agents efficient retrieval may also find the guide to building vector search in Python a useful companion.