RAG Agent
%%{init: {'look': 'handDrawn', 'theme': 'base', 'themeVariables': {'primaryColor': '#eef2ff', 'primaryBorderColor': '#1e40af', 'primaryTextColor': '#1e293b', 'lineColor': '#1e3a8a', 'edgeLabelBackground': '#ffffff', 'clusterBkg': '#fbfcff', 'clusterBorder': '#2563eb', 'fontFamily': '-apple-system, system-ui, Segoe UI, Roboto, Helvetica, Arial, sans-serif', 'fontSize': '15px'}, 'flowchart': {'nodeSpacing': 50, 'rankSpacing': 58, 'padding': 14, 'htmlLabels': true, 'curve': 'basis'}}}%%
flowchart LR
Q(["Question"]) --> S("Search the<br/>knowledge base")
S --> G{"Enough to<br/>answer?"}
G -. "no · try a sharper query" .-> S
G == "yes" ==> A("Answer, with the<br/>sources it used")Outcome: retrieve context for a question, have a model grade whether that context can actually answer it, rewrite the query and retry when it cannot, and refuse to answer when grounding never arrives.
Why the loop matters
A two-step RAG chain — search, then answer — has no idea whether what it retrieved is relevant. The model is handed weak context and a question, and its instructions tell it to answer, so it does. That failure is silent and looks exactly like success.
This recipe splits the two jobs. grade_retrieved_context is a separate call that is explicitly forbidden from answering; it only decides whether the evidence is sufficient and, if not, proposes a better search phrasing. The loop then re-searches with that phrasing. Three outcomes are possible, and all three are recorded:
| Grading result | What happens |
|---|---|
| Sufficient | Answer with citations, then verify at least one citation exists |
| Insufficient, attempts left | Rewrite the query and search again |
| Insufficient after 3 rounds | TERMINATE with insufficient_grounding and the reason |
That third row is the production-relevant one. A workflow that fails loudly is recoverable; one that returns a confident ungrounded answer is not.
Prerequisites
A configured vector database and an OpenAI integration. Index-time and query-time embedding models must match exactly — different embedding spaces produce meaningless similarity scores.
Populate the index before running this. Use LLM_INDEX_TEXT with a stable docId and a metadata object per document, so the citations this workflow returns point at something you can resolve later:
{
"name": "index_policy_doc",
"taskReferenceName": "index_policy_doc",
"type": "LLM_INDEX_TEXT",
"inputParameters": {
"vectorDB": "REPLACE_VECTOR_DB",
"index": "REPLACE_INDEX",
"namespace": "REPLACE_NAMESPACE",
"docId": "retention-policy-v4",
"text": "REPLACE with the document body",
"embeddingModelProvider": "openai",
"embeddingModel": "text-embedding-3-small",
"dimensions": 1536,
"metadata": { "sourceVersion": "v4", "category": "policy" }
}
}
Keep ingestion in its own workflow. Re-indexing on every question wastes embedding spend and makes the answer path depend on write availability.
Runnable definition
Save this as rag-agent.json:
{
"name": "rag_agent",
"description": "Grounded RAG with a bounded retrieval-refinement loop. Retrieves context, grades whether it can actually answer, rewrites the query and retries when it cannot, and refuses to answer rather than answering ungrounded.",
"version": 1,
"schemaVersion": 2,
"timeoutSeconds": 600,
"timeoutPolicy": "TIME_OUT_WF",
"inputParameters": [
"question",
"vectorDB",
"index",
"namespace"
],
"variables": {
"search_query": "",
"sources": [],
"grounding": {
"sufficient": false,
"reason": "not_attempted",
"attempts": 0
}
},
"tasks": [
{
"name": "seed_search_query",
"taskReferenceName": "seed_query",
"type": "SET_VARIABLE",
"inputParameters": {
"search_query": "${workflow.input.question}"
}
},
{
"name": "retrieve_and_grade_loop",
"taskReferenceName": "retrieval_loop",
"type": "DO_WHILE",
"evaluatorType": "graaljs",
"inputParameters": {
"retrieval_loop": "${retrieval_loop.output}",
"sufficient": "${grade_context.output.result.sufficient}"
},
"loopCondition": "(function(){ return $.retrieval_loop['iteration'] < 3 && $.sufficient !== true; })();",
"loopOver": [
{
"name": "retrieve_sources",
"taskReferenceName": "retrieve_sources",
"type": "LLM_SEARCH_INDEX",
"inputParameters": {
"vectorDB": "${workflow.input.vectorDB}",
"index": "${workflow.input.index}",
"namespace": "${workflow.input.namespace}",
"embeddingModelProvider": "openai",
"embeddingModel": "text-embedding-3-small",
"dimensions": 1536,
"query": "${workflow.variables.search_query}",
"maxResults": 5
}
},
{
"name": "grade_retrieved_context",
"taskReferenceName": "grade_context",
"type": "LLM_CHAT_COMPLETE",
"inputParameters": {
"llmProvider": "openai",
"model": "gpt-4o-mini",
"messages": [
{
"role": "system",
"message": "You grade retrieval quality. You do NOT answer the question. Decide only whether the supplied sources contain enough specific evidence to answer it fully. Return JSON: {\"sufficient\": boolean, \"reason\": string, \"refinedQuery\": string}. If sufficient is false, refinedQuery must be a different, more targeted search phrasing that would find the missing evidence. Prefer sufficient=false when the sources are only tangentially related."
},
{
"role": "user",
"message": "Question: ${workflow.input.question}\nCurrent search query: ${workflow.variables.search_query}\nRetrieved sources: ${retrieve_sources.output.result}"
}
],
"temperature": 0.0,
"maxTokens": 500,
"jsonOutput": true
}
},
{
"name": "record_grounding_state",
"taskReferenceName": "record_grounding",
"type": "SET_VARIABLE",
"inputParameters": {
"search_query": "${grade_context.output.result.refinedQuery}",
"sources": "${retrieve_sources.output.result}",
"grounding": {
"sufficient": "${grade_context.output.result.sufficient}",
"reason": "${grade_context.output.result.reason}",
"attempts": "${retrieval_loop.output.iteration}"
}
}
}
]
},
{
"name": "route_on_grounding",
"taskReferenceName": "route_grounding",
"type": "SWITCH",
"evaluatorType": "value-param",
"expression": "sufficient",
"inputParameters": {
"sufficient": "${workflow.variables.grounding.sufficient}"
},
"decisionCases": {
"true": [
{
"name": "answer_from_sources",
"taskReferenceName": "answer",
"type": "LLM_CHAT_COMPLETE",
"inputParameters": {
"llmProvider": "openai",
"model": "gpt-4o-mini",
"messages": [
{
"role": "system",
"message": "Answer strictly from the supplied sources. Return JSON: {\"answer\": string, \"citations\": [string]}. Every citation must be a document id present in the sources. If a claim is not supported by a source, omit the claim. Never use outside knowledge."
},
{
"role": "user",
"message": "Question: ${workflow.input.question}\nSources: ${workflow.variables.sources}"
}
],
"temperature": 0.1,
"maxTokens": 900,
"jsonOutput": true
}
},
{
"name": "verify_citations_present",
"taskReferenceName": "verify_citations",
"type": "JSON_JQ_TRANSFORM",
"inputParameters": {
"answer": "${answer.output.result}",
"queryExpression": "{cited: ((.answer.citations // []) | length), grounded: (((.answer.citations // []) | length) > 0)}"
}
}
],
"false": [
{
"name": "refuse_ungrounded_answer",
"taskReferenceName": "refuse",
"type": "TERMINATE",
"inputParameters": {
"terminationStatus": "FAILED",
"workflowOutput": {
"error": "insufficient_grounding",
"detail": "${workflow.variables.grounding.reason}",
"attempts": "${workflow.variables.grounding.attempts}"
}
}
}
]
},
"defaultCase": []
}
],
"outputParameters": {
"answer": "${answer.output.result.answer}",
"citations": "${answer.output.result.citations}",
"citationCount": "${verify_citations.output.result.cited}",
"grounding": "${workflow.variables.grounding}",
"sources": "${workflow.variables.sources}"
}
}
Register and run
conductor workflow create rag-agent.json
conductor workflow start -w rag_agent --sync -i '{"question":"What is our data retention policy?","vectorDB":"REPLACE_VECTOR_DB","index":"REPLACE_INDEX","namespace":"REPLACE_NAMESPACE"}'
Open Executions in the Conductor UI and select the new execution to review the task graph, and each task's inputs and outputs.
Look at how many times retrieval_loop iterated. One iteration means the first query was good enough. Three plus a FAILED status means your index does not contain the answer — which is a real, useful signal about your corpus rather than a workflow bug.
Production notes
maxResultsdefaults to 1. Get the name wrong and you silently retrieve one document, which looks like a bad retriever.- Grade with a cheap model, answer with a good one. Grading runs up to three times per question, so it drives the cost.
- Treat citations as a contract. Reject answers whose citations don't resolve against your index rather than showing them.
- Index once, in its own workflow. Re-indexing per question wastes embedding spend and couples answering to write availability.
- Match the embedding model at index and query time. Different embedding spaces make similarity scores meaningless.
- Cache on the question plus index version so a re-indexed corpus invalidates it.