{"id":867,"date":"2026-05-07T07:00:03","date_gmt":"2026-05-06T23:00:03","guid":{"rendered":"https:\/\/connectword.dpdns.org\/?p=867"},"modified":"2026-05-07T07:00:03","modified_gmt":"2026-05-06T23:00:03","slug":"a-groq-powered-agentic-research-assistant-with-langgraph-tool-calling-sub-agents-and-agentic-memory-lets-built-it","status":"publish","type":"post","link":"https:\/\/connectword.dpdns.org\/?p=867","title":{"rendered":"A Groq-Powered Agentic Research Assistant with LangGraph, Tool Calling, Sub-Agents, and Agentic Memory: Lets Built It"},"content":{"rendered":"

In this tutorial, we build a Groq<\/strong><\/a>-powered agentic research workflow that runs directly using Groq\u2019s free OpenAI-compatible inference endpoint. We configure LangChain\u2019s ChatOpenAI interface to work with Groq by setting the Groq API key and base URL, allowing us to use fast hosted models such as llama-3.3-70b-versatile for tool-based reasoning. We then connect the model with practical tools for web search, webpage fetching, file handling, Python execution, skill loading, sub-agent delegation, and long-term memory. By the end of the tutorial, we have a working Groq-based multi-step agent that can research a topic, delegate focused subtasks, generate structured outputs, and save useful information for later runs.<\/p>\n

\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
import subprocess, sys\ndef _pip(*a): subprocess.check_call([sys.executable,\"-m\",\"pip\",\"install\",\"-q\",*a])\n_pip(\"langgraph>=0.2.50\", \"langchain>=0.3.0\", \"langchain-openai>=0.2.0\",\n    \"langchain-community>=0.3.0\", \"ddgs\", \"requests\", \"beautifulsoup4\",\n    \"tiktoken\", \"pydantic>=2.0\")\n\n\nimport os, getpass\nif not os.environ.get(\"GROQ_API_KEY\"):\n   os.environ[\"GROQ_API_KEY\"] = getpass.getpass(\"GROQ_API_KEY (free at console.groq.com\/keys): \")\n\n\nos.environ[\"OPENAI_API_KEY\"]  = os.environ[\"GROQ_API_KEY\"]\nos.environ[\"OPENAI_BASE_URL\"] = \"https:\/\/api.groq.com\/openai\/v1\"\n\n\nMODEL_NAME = \"llama-3.3-70b-versatile\"\n\n\nimport json, re, io, contextlib, pathlib\nfrom typing import Annotated, TypedDict, Sequence, Literal, List, Dict, Any\nfrom datetime import datetime, timezone\nfrom langchain_openai import ChatOpenAI\nfrom langchain_core.messages import (\n   SystemMessage, HumanMessage, AIMessage, ToolMessage, BaseMessage)\nfrom langchain_core.tools import tool\nfrom langgraph.graph import StateGraph, END\nfrom langgraph.graph.message import add_messages\nfrom langgraph.prebuilt import ToolNode<\/code><\/pre>\n<\/div>\n<\/div>\n
We install the core libraries required to build the Groq-powered agent workflow, including LangGraph, LangChain, DuckDuckGo search utilities, and supporting parsing libraries. We securely collect the Groq API key and configure Groq as an OpenAI-compatible endpoint by setting the API key and base URL. We then import all required modules for messages, tools, graph construction, typing, filesystem handling, and model initialization.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
SANDBOX = pathlib.Path(\"\/content\/deerflow_sandbox\").resolve()\nfor sub in [\"uploads\",\"workspace\",\"outputs\",\"skills\/public\",\"skills\/custom\",\"memory\"]:\n   (SANDBOX\/sub).mkdir(parents=True, exist_ok=True)\n\n\ndef _safe(p: str) -> pathlib.Path:\n   full = (SANDBOX\/p.lstrip(\"\/\")).resolve()\n   if not str(full).startswith(str(SANDBOX)):\n       raise ValueError(f\"path escapes sandbox: {p}\")\n   return full\n\n\nSKILLS: Dict[str, Dict[str,str]] = {}\ndef register_skill(name, description, content, location=\"public\"):\n   d = SANDBOX\/\"skills\"\/location\/name; d.mkdir(parents=True, exist_ok=True)\n   (d\/\"SKILL.md\").write_text(content)\n   SKILLS[name] = {\"description\": description, \"content\": content,\n                   \"path\": str(d\/\"SKILL.md\")}\n\n\nregister_skill(\"research\",\n   \"Conduct multi-source web research on a topic and produce structured notes.\",\n   \"\"\"# Research Skill\n## Workflow\n1. Decompose the question into 3-5 sub-questions.\n2. For each sub-question call `web_search` and pick 2 authoritative URLs.\n3. `web_fetch` those URLs; extract concrete facts, numbers, dates.\n4. Cross-reference for consensus vs. disagreement.\n5. Append findings to `workspace\/research_notes.md`: claim \u2192 evidence \u2192 URL.\n## Best practices\n- Prefer primary sources. Note dates. Never fabricate URLs or numbers.\"\"\")\n\n\nregister_skill(\"report-generation\",\n   \"Synthesize research notes into a polished markdown report in outputs\/.\",\n   \"\"\"# Report Generation Skill\n## Workflow\n1. file_read('workspace\/research_notes.md').\n2. Outline: exec summary, key findings, analysis, conclusion, sources.\n3. file_write('outputs\/report.md', ...).\n## Structure\n- # Title\n- ## Executive Summary  (3\u20135 sentences)\n- ## Key Findings       (bullets)\n- ## Detailed Analysis  (sections)\n- ## Conclusion\n- ## Sources            (numbered URL list)\"\"\")\n\n\nregister_skill(\"code-execution\",\n   \"Run Python in the sandbox for computation, data wrangling, charts.\",\n   \"\"\"# Code Execution Skill\n1. Plan in plain language first.\n2. python_exec the code; persistent artifacts go to \/outputs\/.\n3. Verify before quoting results.\"\"\")\n\n\nMEM = SANDBOX\/\"memory\/long_term.json\"\nif not MEM.exists():\n   MEM.write_text(json.dumps({\"facts\":[],\"preferences\":{}}, indent=2))\ndef _load_mem(): return json.loads(MEM.read_text())\ndef _save_mem(m): MEM.write_text(json.dumps(m, indent=2))<\/code><\/pre>\n<\/div>\n<\/div>\n
We create a sandboxed project directory in Colab to keep uploads, workspace files, outputs, skills, and memory organized in a single controlled location. We define reusable skills for research, report generation, and code execution so the agent can discover and follow structured workflows. We also initialize a simple long-term memory JSON file that stores facts and preferences across multiple runs within the same sandbox.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
@tool\ndef list_skills() -> str:\n   \"\"\"List all skills with one-line descriptions. Call this first for complex tasks.\"\"\"\n   return \"n\".join(f\"- {n}: {s['description']}\" for n,s in SKILLS.items())\n\n\n@tool\ndef load_skill(name: str) -> str:\n   \"\"\"Load full SKILL.md for `name`. Call before running its workflow.\"\"\"\n   if name not in SKILLS: return f\"Unknown. Available: {list(SKILLS)}\"\n   return SKILLS[name][\"content\"]\n\n\n@tool\ndef web_search(query: str, max_results: int = 5) -> str:\n   \"\"\"Search the web (DuckDuckGo). Returns titles, URLs, snippets.\"\"\"\n   from ddgs import DDGS\n   out = []\n   try:\n       with DDGS() as d:\n           for r in d.text(query, max_results=max_results):\n               out.append(f\"- {r.get('title','')}n  URL: {r.get('href','')}n  \"\n                          f\"{(r.get('body') or '')[:220]}\")\n   except Exception as e:\n       return f\"search error: {e}\"\n   return \"n\".join(out) or \"no results\"\n\n\n@tool\ndef web_fetch(url: str, max_chars: int = 4000) -> str:\n   \"\"\"Fetch a URL, return cleaned text (scripts\/nav stripped).\"\"\"\n   import requests\n   from bs4 import BeautifulSoup\n   try:\n       r = requests.get(url, timeout=15,\n                        headers={\"User-Agent\":\"Mozilla\/5.0 DeerFlow-Lite\"})\n       soup = BeautifulSoup(r.text, \"html.parser\")\n       for s in soup([\"script\",\"style\",\"nav\",\"footer\",\"aside\",\"header\"]): s.decompose()\n       text = re.sub(r\"ns*n\", \"nn\", soup.get_text(\"n\")).strip()\n       return text[:max_chars] or \"(empty page)\"\n   except Exception as e:\n       return f\"fetch error: {e}\"\n\n\n@tool\ndef file_write(path: str, content: str) -> str:\n   \"\"\"Write content to a sandbox path, e.g. 'workspace\/notes.md' or 'outputs\/x.md'.\"\"\"\n   p = _safe(path); p.parent.mkdir(parents=True, exist_ok=True)\n   p.write_text(content)\n   return f\"wrote {len(content)} chars \u2192 {path}\"\n\n\n@tool\ndef file_read(path: str) -> str:\n   \"\"\"Read a sandbox file (first 8 KB).\"\"\"\n   p = _safe(path)\n   return p.read_text()[:8000] if p.exists() else f\"not found: {path}\"\n\n\n@tool\ndef file_list(path: str = \"\") -> str:\n   \"\"\"List files under a sandbox dir.\"\"\"\n   base = _safe(path) if path else SANDBOX\n   if not base.exists(): return \"not found\"\n   items = []\n   for c in sorted(base.rglob(\"*\")):\n       if \"memory\" in c.relative_to(SANDBOX).parts: continue\n       items.append(f\"  {'D' if c.is_dir() else 'F'}  {c.relative_to(SANDBOX)}\")\n   return \"n\".join(items[:60]) or \"(empty)\"\n\n\n@tool\ndef python_exec(code: str) -> str:\n   \"\"\"Run Python in the sandbox. SANDBOX_ROOT is preset.\"\"\"\n   g = {\"__name__\":\"__sb__\", \"SANDBOX_ROOT\": str(SANDBOX)}\n   buf = io.StringIO()\n   try:\n       with contextlib.redirect_stdout(buf), contextlib.redirect_stderr(buf):\n           exec(code, g)\n       return (buf.getvalue() or \"(no stdout)\")[:4000]\n   except Exception as e:\n       return f\"{type(e).__name__}: {e}n{buf.getvalue()[:1500]}\"\n\n\n@tool\ndef remember(fact: str) -> str:\n   \"\"\"Persist a single fact to long-term memory (survives across runs).\"\"\"\n   m = _load_mem()\n   m[\"facts\"].append({\"fact\": fact, \"ts\": datetime.now(timezone.utc).isoformat()})\n   _save_mem(m)\n   return f\"remembered ({len(m['facts'])} total)\"\n\n\n@tool\ndef recall() -> str:\n   \"\"\"Retrieve everything in long-term memory.\"\"\"\n   m = _load_mem()\n   if not m[\"facts\"]: return \"(memory empty)\"\n   return \"n\".join(f\"- {f['fact']}\" for f in m[\"facts\"][-20:])<\/code><\/pre>\n<\/div>\n<\/div>\n
We define the main tools the Groq-backed agent can call during execution, including listing skills, loading skill instructions, searching the web, fetching webpages, reading files, and writing files. We also provide the agent with a sandboxed Python execution environment so it can run computations or generate artifacts when needed. We add memory tools that allow the agent to remember important facts and recall previously stored information.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
@tool\ndef spawn_subagent(role: str, task: str,\n                  allowed_tools: str = \"web_search,web_fetch,file_write,file_read\") -> str:\n   \"\"\"Spawn an isolated sub-agent with a focused role and scoped tools.\n   Returns its final report string. Use for parallelizable \/ focused subtasks.\"\"\"\n   bag = {t.name: t for t in BASE_TOOLS}\n   sub_tools = [bag[n.strip()] for n in allowed_tools.split(\",\") if n.strip() in bag]\n   sub_llm = ChatOpenAI(model=MODEL_NAME, temperature=0.2).bind_tools(sub_tools)\n   sys_msg = SystemMessage(content=(\n       f\"You are a specialized sub-agent. Role: {role}.n\"\n       f\"You operate in an ISOLATED context \u2014 no access to lead history.n\"\n       f\"Tools: {', '.join(t.name for t in sub_tools)}.n\"\n       \"End with a final assistant message starting 'FINAL REPORT:' \"\n       \"containing a structured \u2264700-word summary including any URLs.\"))\n   msgs: List[BaseMessage] = [sys_msg, HumanMessage(content=task)]\n   for _ in range(8):\n       r = sub_llm.invoke(msgs); msgs.append(r)\n       if not getattr(r, \"tool_calls\", None):\n           return f\"[sub-agent: {role}]n\" + (r.content if isinstance(r.content,str) else str(r.content))\n       for tc in r.tool_calls:\n           t = bag.get(tc[\"name\"])\n           try:\n               res = t.invoke(tc[\"args\"]) if t else f\"unknown tool {tc['name']}\"\n           except Exception as e:\n               res = f\"tool error: {e}\"\n           msgs.append(ToolMessage(content=str(res)[:3000], tool_call_id=tc[\"id\"]))\n   return f\"[sub-agent: {role}] step-limit reached.\"\n\n\nBASE_TOOLS = [list_skills, load_skill, web_search, web_fetch, file_write,\n             file_read, file_list, python_exec, remember, recall]\nALL_TOOLS = BASE_TOOLS + [spawn_subagent]\n\n\nLEAD_SYSTEM = f\"\"\"You are DeerFlow-Lite, a long-horizon super-agent harness.\n\n\nSandbox layout (relative to {SANDBOX}):\n uploads\/    \u2013 user files\n workspace\/  \u2013 your scratchpad\n outputs\/    \u2013 final deliverables\n skills\/     \u2013 capability modules (load_skill)\n\n\nPrinciples:\n \u2022 For non-trivial tasks: list_skills \u2192 load_skill \u2192 execute.\n \u2022 Use spawn_subagent for focused subtasks (isolated context keeps lead lean).\n \u2022 Persist intermediates to workspace\/, deliverables to outputs\/.\n \u2022 Use remember(fact) for cross-session knowledge.\n \u2022 Finish with a short summary of what was produced and where.\n\n\nToday: {datetime.now(timezone.utc).strftime('%Y-%m-%d')}.\"\"\"\n\n\nclass AgentState(TypedDict):\n   messages: Annotated[Sequence[BaseMessage], add_messages]\n\n\nllm = ChatOpenAI(model=MODEL_NAME, temperature=0.3).bind_tools(ALL_TOOLS)\n\n\ndef call_model(state: AgentState):\n   msgs = list(state[\"messages\"])\n   if not msgs or not isinstance(msgs[0], SystemMessage):\n       msgs = [SystemMessage(content=LEAD_SYSTEM)] + msgs\n   return {\"messages\": [llm.invoke(msgs)]}\n\n\ndef route(state: AgentState) -> Literal[\"tools\",\"__end__\"]:\n   last = state[\"messages\"][-1]\n   return \"tools\" if getattr(last, \"tool_calls\", None) else END\n\n\ng = StateGraph(AgentState)\ng.add_node(\"agent\", call_model)\ng.add_node(\"tools\", ToolNode(ALL_TOOLS))\ng.set_entry_point(\"agent\")\ng.add_conditional_edges(\"agent\", route, {\"tools\":\"tools\", END: END})\ng.add_edge(\"tools\", \"agent\")\nAPP = g.compile()<\/code><\/pre>\n<\/div>\n<\/div>\n
We create a sub-agent tool that allows the main Groq-powered agent to delegate focused tasks to an isolated assistant with a limited set of tools. We then collect all available tools, define the lead system prompt, initialize the Groq-backed chat model, and bind the tools to it. We finally built the LangGraph workflow so the agent can alternate between reasoning and tool execution until it reaches a final answer.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
def run(task: str, max_steps: int = 25):\n   print(\"=\"*78); print(f\"\"\ud83e\udd8c\" TASK: {task}\"); print(\"=\"*78)\n   state = {\"messages\":[HumanMessage(content=task)]}\n   n = 0\n   for ev in APP.stream(state, {\"recursion_limit\": max_steps*2}, stream_mode=\"updates\"):\n       for node, payload in ev.items():\n           for m in payload.get(\"messages\", []):\n               n += 1\n               if isinstance(m, AIMessage):\n                   if m.tool_calls:\n                       for tc in m.tool_calls:\n                           args = json.dumps(tc[\"args\"], ensure_ascii=False)\n                           args = args[:140] + (\"\u2026\" if len(args)>140 else \"\")\n                           print(f\"[{n:02}] \"\ud83d\udd27\" {tc['name']}({args})\")\n                   else:\n                       txt = m.content if isinstance(m.content,str) else str(m.content)\n                       print(f\"[{n:02}] \"\ud83e\udd8c\" {txt[:800]}\")\n               elif isinstance(m, ToolMessage):\n                   s = str(m.content).replace(\"n\",\" \")[:220]\n                   print(f\"[{n:02}] \"\ud83d\udce4\" {s}\")\n   print(\"n\"+\"=\"*78); print(\"\"\u2705\" COMPLETE \u2014 sandbox state:\"); print(\"=\"*78)\n   print(file_list.invoke({\"path\":\"\"}))\n   print(\"n\"\ud83e\udde0\" Long-term memory:\"); print(recall.invoke({}))\n   for f in sorted((SANDBOX\/\"outputs\").rglob(\"*\")):\n       if f.is_file():\n           print(f\"n--- \"\ud83d\udcc4\" {f.relative_to(SANDBOX)} (first 800 chars) ---\")\n           print(f.read_text()[:800])\n\n\nrun(\n   \"Give me a briefing on small language models (SLMs) in 2025. \"\n   \"(1) discover skills; (2) spawn a researcher sub-agent to gather \"\n   \"specifics on three notable SLMs from 2024-2025 with sizes, benchmarks, \"\n   \"and use cases \u2014 sub-agent saves to workspace\/slm_research.md; \"\n   \"(3) load report-generation skill and write outputs\/slm_briefing.md \"\n   \"(~400 words) with a Sources section; (4) save the single most \"\n   \"important takeaway to long-term memory; (5) summarize.\",\n   max_steps=25,\n)\n<\/code><\/pre>\n<\/div>\n<\/div>\n
We define the run() function that starts a user task, streams each agent step, and prints tool calls, tool outputs, and final responses in a readable format. We also display the sandbox file structure, long-term memory, and generated output files after the workflow completes. We finish by running a demo task in which the Groq-powered agent researches small language models, prepares a briefing, saves a report, and stores one key takeaway in memory.<\/p>\n
In conclusion, we created a compact yet capable Groq-based agent framework that demonstrates how Groq\u2019s OpenAI-compatible API can serve as a fast, accessible backend for advanced LLM workflows. We used LangGraph to manage the agent loop, LangChain to bind tools to the Groq-hosted model, and custom Python utilities to give the system controlled access to search, files, code execution, and memory. We also demonstrated how isolated sub-agents can help handle focused research tasks while the main agent coordinates the overall workflow. Also, we finished with a practical Groq-powered agentic system that can be extended into research assistants, automated briefing generators, and multi-step AI applications.<\/p>\n
\n
Check out\u00a0the\u00a0Full Codes with Notebook here<\/a><\/strong>.\u00a0<\/strong>Also,\u00a0feel free to follow us on\u00a0Twitter<\/mark><\/a><\/strong>\u00a0and don\u2019t forget to join our\u00a0130k+ ML SubReddit<\/a><\/strong>\u00a0and Subscribe to\u00a0our Newsletter<\/a><\/strong>. Wait! are you on telegram?\u00a0now you can join us on telegram as well.<\/a><\/strong><\/p>\n
Need to partner with us for promoting your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar etc.?\u00a0Connect with us<\/mark><\/a><\/strong><\/p>\n
The post A Groq-Powered Agentic Research Assistant with LangGraph, Tool Calling, Sub-Agents, and Agentic Memory: Lets Built It<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"
In this tutorial, we build a G…<\/p>\n","protected":false},"author":1,"featured_media":29,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-867","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/867","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=867"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/867\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/media\/29"}],"wp:attachment":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=867"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=867"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=867"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}