{"title":"Exercise: Agentic RAG","description":null,"content":"# Exercise 1: Route a Query Before Retrieving\n\nBuild a router that decides how to handle each incoming query before doing any retrieval: answer directly when the question is general knowledge or chit-chat, and retrieve from the knowledge base only when the question is about the specific documents. Then act on that decision.\n\nStandard RAG retrieves for every query, which wastes a retrieval call on \"hello\" and can pollute the context with irrelevant chunks for questions the model could answer on its own. Agentic RAG starts by letting the model classify the query and choose an action. This routing step is the simplest form of agency in a RAG system, and it is what keeps retrieval focused on the questions that actually need it.\n\n### Requirements\n\n- Ask the model to classify the query as `retrieve` (needs the knowledge base) or `direct` (answer without retrieval).\n- If the decision is `retrieve`, select the most similar chunk from the corpus and answer from it.\n- If the decision is `direct`, answer without using the corpus.\n- Print the decision and the answer for each query.\n\n\n**main.py**\n\n```python\nfrom openai import OpenAI\nimport math\n\nclient = OpenAI() # Uses OPENAI_API_KEY from environment\n\ncorpus = [\n \"Our API rate limit is 100 requests per minute on the free tier.\",\n \"Webhooks retry up to 5 times with exponential backoff before giving up.\",\n \"API keys can be rotated from the dashboard under Settings, Security.\",\n]\n\nqueries = [\n \"What is the rate limit on the free tier?\",\n \"Thanks, that was helpful!\",\n]\n\ndef cosine_similarity(a, b):\n dot = sum(x * y for x, y in zip(a, b))\n return dot / (math.sqrt(sum(x * x for x in a)) * math.sqrt(sum(y * y for y in b)))\n\ndef route(query: str) -> str:\n # TODO: ask the model to classify the query as \"retrieve\" or \"direct\".\n # Return the lowercased decision string.\n pass\n\ndef answer(query: str) -> str:\n decision = route(query)\n # TODO: if decision is \"retrieve\", find the most similar corpus chunk\n # and answer from it. Otherwise, answer directly without the corpus.\n pass\n\n# TODO: for each query, print the routing decision and the answer\n```\n\n\n### Expected Output\n\nThe knowledge-base question is routed to retrieval; the thank-you message is answered directly.\n\n\n```plaintext\n[retrieve] What is the rate limit on the free tier?\n -> The free tier allows 100 requests per minute.\n\n[direct] Thanks, that was helpful!\n -> You're welcome! Happy to help if anything else comes up.\n```\n\n\n### Solution\n\n\n**main.py**\n\n```python\nfrom openai import OpenAI\nimport math\n\nclient = OpenAI()\n\ncorpus = [\n \"Our API rate limit is 100 requests per minute on the free tier.\",\n \"Webhooks retry up to 5 times with exponential backoff before giving up.\",\n \"API keys can be rotated from the dashboard under Settings, Security.\",\n]\n\nqueries = [\n \"What is the rate limit on the free tier?\",\n \"Thanks, that was helpful!\",\n]\n\ndef cosine_similarity(a, b):\n dot = sum(x * y for x, y in zip(a, b))\n return dot / (math.sqrt(sum(x * x for x in a)) * math.sqrt(sum(y * y for y in b)))\n\ndef route(query: str) -> str:\n response = client.chat.completions.create(\n model=\"gpt-5.4-mini\",\n messages=[\n {\n \"role\": \"system\",\n \"content\": (\n \"Classify the query. Reply with exactly one word: 'retrieve' if it \"\n \"asks about a specific product, API, or internal documentation, or \"\n \"'direct' if it is general knowledge, chit-chat, or a greeting.\"\n ),\n },\n {\"role\": \"user\", \"content\": query},\n ],\n )\n return response.choices[0].message.content.strip().lower()\n\ndef answer(query: str) -> str:\n decision = route(query)\n if decision == \"retrieve\":\n query_vector = client.embeddings.create(\n model=\"text-embedding-3-small\", input=query\n ).data[0].embedding\n corpus_vectors = [\n item.embedding\n for item in client.embeddings.create(model=\"text-embedding-3-small\", input=corpus).data\n ]\n best = max(range(len(corpus)), key=lambda i: cosine_similarity(query_vector, corpus_vectors[i]))\n context = corpus[best]\n messages = [\n {\"role\": \"system\", \"content\": \"Answer using only the provided context.\"},\n {\"role\": \"user\", \"content\": f\"Context: {context}\\n\\nQuestion: {query}\"},\n ]\n else:\n messages = [{\"role\": \"user\", \"content\": query}]\n\n response = client.chat.completions.create(model=\"gpt-5.4-mini\", messages=messages)\n return decision, response.choices[0].message.content.strip()\n\nfor query in queries:\n decision, reply = answer(query)\n print(f\"[{decision}] {query}\")\n print(f\" -> {reply}\\n\")\n```\n\n\nThe router is a small classification call that runs before any retrieval, turning \"always retrieve\" into \"retrieve when it helps.\" That decision changes the control flow: a `retrieve` query embeds and searches the corpus, while a `direct` query skips straight to generation. Extending this idea (more actions, multiple retrieval sources, a loop that re-routes after seeing results) is what grows a RAG pipeline into an agent.\n\n---\n\n# Exercise 2: A Three-Way Router with Clarification\n\nA two-way retrieve-or-answer router still struggles with vague input. Add a third route: when a query is too ambiguous to act on, the agent asks for clarification instead of guessing. This is the difference between an agent that confidently answers the wrong question and one that knows when to ask.\n\n### Requirements\n\n- Classify each query as `retrieve`, `direct`, or `clarify`.\n- For `retrieve`, fetch from the corpus and answer; for `direct`, answer without the corpus; for `clarify`, return a clarifying question.\n- Print the route and the response for each query.\n\n\n**main.py**\n\n```python\nimport math\nfrom openai import OpenAI\n\nclient = OpenAI() # Uses OPENAI_API_KEY from environment\n\ncorpus = [\n \"Our API rate limit is 100 requests per minute on the free tier.\",\n \"Webhooks retry up to 5 times with exponential backoff.\",\n]\n\nqueries = [\n \"What is the rate limit?\",\n \"Thanks for your help!\",\n \"Can you help me with it?\",\n]\n\ndef cosine_similarity(a, b):\n dot = sum(x * y for x, y in zip(a, b))\n return dot / (math.sqrt(sum(x * x for x in a)) * math.sqrt(sum(y * y for y in b)))\n\ndef route(query):\n # TODO: classify as \"retrieve\" (needs the docs), \"direct\" (chit-chat/general),\n # or \"clarify\" (too vague to act on). Return the lowercased label.\n ...\n\n# TODO: for each query, route it and act: retrieve+answer, answer directly,\n# or ask a clarifying question. Print the route and the response.\n```\n\n\n### Expected Output\n\nThe factual query is retrieved, the thank-you is answered directly, and the vague \"help me with it\" triggers a clarifying question.\n\n\n```plaintext\n[retrieve] What is the rate limit?\n -> The free tier allows 100 requests per minute.\n\n[direct] Thanks for your help!\n -> You're welcome! Glad to help.\n\n[clarify] Can you help me with it?\n -> Could you clarify what you are referring to?\n\n```\n\n\n### Solution\n\n\n**main.py**\n\n```python\nimport math\nfrom openai import OpenAI\n\nclient = OpenAI() # Uses OPENAI_API_KEY from environment\n\ncorpus = [\n \"Our API rate limit is 100 requests per minute on the free tier.\",\n \"Webhooks retry up to 5 times with exponential backoff.\",\n]\n\nqueries = [\n \"What is the rate limit?\",\n \"Thanks for your help!\",\n \"Can you help me with it?\",\n]\n\ndef cosine_similarity(a, b):\n dot = sum(x * y for x, y in zip(a, b))\n return dot / (math.sqrt(sum(x * x for x in a)) * math.sqrt(sum(y * y for y in b)))\n\ndef route(query):\n response = client.chat.completions.create(\n model=\"gpt-5.4-mini\",\n messages=[{\"role\": \"system\", \"content\": \"Classify the query as one word: 'retrieve' if it asks about the product/API, 'direct' if it is chit-chat or general knowledge, or 'clarify' if it is too vague to answer.\"},\n {\"role\": \"user\", \"content\": query}],\n ).choices[0].message.content.strip().lower()\n return response\n\ndef handle(query):\n decision = route(query)\n if decision == \"retrieve\":\n qv = client.embeddings.create(model=\"text-embedding-3-small\", input=query).data[0].embedding\n cv = [item.embedding for item in client.embeddings.create(model=\"text-embedding-3-small\", input=corpus).data]\n context = corpus[max(range(len(corpus)), key=lambda i: cosine_similarity(qv, cv[i]))]\n msgs = [{\"role\": \"system\", \"content\": \"Answer using only the context.\"},\n {\"role\": \"user\", \"content\": f\"Context: {context}\\n\\nQuestion: {query}\"}]\n elif decision == \"clarify\":\n return decision, \"Could you clarify what you are referring to?\"\n else:\n msgs = [{\"role\": \"user\", \"content\": query}]\n reply = client.chat.completions.create(model=\"gpt-5.4-mini\", messages=msgs).choices[0].message.content.strip()\n return decision, reply\n\nfor query in queries:\n decision, reply = handle(query)\n print(f\"[{decision}] {query}\\n -> {reply}\\n\")\n```\n\n\nThe third route encodes a basic competence: recognizing when there is not enough information to act. A vague query routed to `retrieve` would fetch an arbitrary chunk and answer confidently about the wrong thing; routing it to `clarify` keeps the agent honest. Adding routes this way is how an agentic system grows, each new action handles a case the simpler router got wrong, and the router stays a cheap classification call in front of the expensive work.","pageType":"ai-engineering"}

Get Premium