Project Goals
- Building an async multi-agent system in LangGraph — supervisor routes each turn to an enhancer, coder, or researcher, with a validator gate and human-in-the-loop before returning control.
- Writing the retrieval layer as a self-correcting RAG sub-agent exposed as a tool: hybrid search (Chroma dense + BM25, weighted rank fusion) → Cohere reranking → per-chunk LLM relevance grading, with automatic query refinement on failed retrieval and an explicit "cannot answer" path instead of a hallucinated one.
- Grading through modulation of the cross-encoder confidence thresholds on epistemic uncertainty enables the agent to recognize knowledge gaps before executing tool calls.
Hard to understand these nerdy technical jargons? No worries bruh. Let's understand each of these with butter-smooth efficiency.
Project Explanation
Line 1 - async multi-agent system in LangGraph
In this system, a supervisor reads each human message and routes it to a specialist. These specialists are : enhancer (fixes vague queries), coder (runs code/math), or researcher (finds facts). A validator checks the answer before it reaches you — good → back to you, bad → back to supervisor. Async = heavy steps run concurrently, cutting latency.
Line 2 - self-correcting RAG sub-agent (used as a tool)
For this, we used Hybrid search, that is "Vector Search" (Chroma) + "Keyword Search" (BM25) fused together, then Cohere takes in the query & retrieved files from Hybrid Search and reranks the top candidates, then an LLM grades each chunk for "does this actually answer it?". If no document is giving the answer, a refiner rewrites the query and retries for a defined no of MAX ITERATIONS; if it still fails, it says "cannot answer" instead of hallucinating.
Line 3 - Grading through modulation of the cross-encoder confidence thresholds on epistemic uncertainty- enables the agent to recognize knowledge gaps before executing tool calls.
Epistemic uncertainty = The agent doesn't know what it doesn't know. Standard RAG just retrieves and answers regardless.
Here, the cross-encoder's relevance score is used as a confidence signal. Moreover, Dynamic modulation using a lighter AI model checks if the data retrieved is enough or should I refine the query and search the vector-store again for allowed MAX_ITERATIONS. Therefore, instead of hallucinating, the agent recognizes the gap before calling a tool (web search, API, etc.) — so it knows when to reach out for external info vs when to answer from retrieved context.
Libraries Required
### Setup the virtual environment :
1. python3 -m venv .venv
2. source .venv/bin/activate
3. Create a (.env) file in your main directory and add the OpenAI, Groq and Cohere API Keys as:
OPENAI_API_KEY="..." GROQ_API_KEY="..." CO_API_KEY="..."
### The Setup : For Mac (Use pip instead of pip3 in windows)
1. pip3 install langchain-chroma langchain-openai langchain-text-splitters langchain-community python-dotenv langchain-classic langchain-cohere langchain-groq langgraph langchain-core
import os
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_ollama import OllamaEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader, DirectoryLoader
from dotenv import load_dotenv
load_dotenv()
# --------------- #
import re
from langchain_classic.retrievers import EnsembleRetriever, BM25Retriever
# --------------- #
from langchain_cohere import CohereRerank
# --------------- #
from langchain_groq import ChatGroq
from typing import List, Annotated, Optional, Literal
from pydantic import BaseModel, Field
from langgraph.types import Command
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage, AIMessage, RemoveMessage
# --------------- #
import asyncio
# --------------- #
import uuid
import time
# --------------- #
embedding_model = OllamaEmbeddings(model='nomic-embed-text')
# embedding_model = OpenAIEmbeddings(model='text-embedding-3-small')
llm = ChatGroq(model = "llama-3.3-70b-versatile")
# --------------- #
from langgraph.graph import StateGraph, END, MessagesState, add_messages
Agent Overview
- The RAG agent is a tool of Researcher node.
RAG Agent : It receives some input from the supervisor node, rewrites the question using chat history to make it RAG searchable. Then it classifies it as if its related to the core topic or not. Off topic and it refuses to answer and returns to supervisor. On topic, it searches top k relevant chunks, grades them according to their relevance to query. If relevant documents are not found, it retries by refining search query for MAX_ITERATIONS defined by us. If still answer can't be found, it reports the supervisor. If answer is found, it returns a well written answer from the retrieved docs.
RAG workflow from query rewriting through retrieval, grading, refinement, and answer generation.
ReAct Agent : Every query given by human is passed on to supervisor (except "exit" or "quit" which ends the agent loop). The supervisor then reviews if it needs to enhance the query to understand it better or is it sufficient. If sufficient, it assigns the task to the required assistant "coder" or "researcher" or answers itself if its a simple query. The result given by the assistants is verified by "validator". If the results are fine, the output is returned and control is passed on to human. If its not, supervisor reviews it and works on it again.
Agent workflow connecting a human, supervisor, enhancer, researcher, coder, and validator.
The Coding Part : RAG Agent
Step1 : A simple setup of a ChromaDB vector store, which loads all the ".txt" files from your "docs" folder and creates a vector store in "db/chroma_db" if it doesn't already exist, else retrieves it.
# Setting up Vector Store
persist_directory = "db/chroma_db"
docs_path="docs"
directory_loader= DirectoryLoader(
path=docs_path,
glob='*.txt',
loader_cls=TextLoader
)
full_docs = directory_loader.load()
text_splitter = RecursiveCharacterTextSplitter(
separators=[r"(?<=\. )", "\n\n", "\n", " ", ""],
is_separator_regex=True,
chunk_size=1000,
chunk_overlap=150)
chunked_docs = text_splitter.split_documents(full_docs)
if not os.path.exists(persist_directory):
print("Creating Vector Store")
vector_store = Chroma.from_documents(
documents=chunked_docs,
persist_directory="db/chroma_db",
embedding=embedding_model,
collection_name="wikipedia_of_companies",
collection_metadata={"hnsw:space" : "cosine"}
)
else:
print("Loading Vector Store")
vector_store = Chroma(
persist_directory=persist_directory,
embedding_function=embedding_model,
collection_name="wikipedia_of_companies",
collection_metadata={"hnsw:space": "cosine"}
)
vector_retriever = vector_store.as_retriever(search_kwargs= {"k" : 10})
# vector_retriever.invoke("What is xlonlabs?")
Step2 : Setting up BM25 Retriever and Hybrid Search Retriever: pip install rank_bm25 before using BM25Retriever
def clean_tokenize(text):
text = text.lower()
text = re.sub(r"(?<!\d)\.(?!\d)|[^\w\s]"," ", text)
return text.split()
bm25_retriever = BM25Retriever.from_documents(documents=chunked_docs, k=10, preprocess_func=clean_tokenize)
# bm25_docs= bm25_retriever.invoke("CEO OF XLONLABS?")
# print(bm25_docs)
hybrid_retriever = EnsembleRetriever(retrievers=[vector_retriever, bm25_retriever], weights=[0.7, 0.3])
# hybrid_search_docs=hybrid_retriever.invoke("ceo of scale")
# print(hybrid_search_docs)
Step3 : Setting up Reranker
reranker = CohereRerank(model='rerank-english-v3.0', top_n=5)
# user_query="Who is the founder of xlon labs?"
# reranked_docs = reranker.compress_documents(hybrid_retriever.invoke(user_query), user_query)
# print(reranked_docs)
Step4 : Setting up Pydantic Models + the "question_rewriter" Node
class RetrievedDocs(BaseModel):
content : str
source : str
relevance_score : float
class AgentState(BaseModel):
user_input : str=""
refined_query : str = ""
docs : List[RetrievedDocs] = Field(default_factory=list)
relevant_docs : List[RetrievedDocs] = Field(default_factory=list)
on_topic : bool = True
retry_count : int = 0
messages : Annotated[List[BaseMessage], add_messages]
def question_rewriter(state: AgentState):
rewriter_prompt = """You are a RAG Agent submodel who rewrites the query and make it RAG Hybrid Search Searchable.
You can look at the previous messages of chat (if available) to make the rewritten query completely searchable.
DON'T CHANGE THE QUESTION ACCORING TO HISTORY FORCEFULLY.
Example :
User - I was searching for who is the ceo of xlonlabs
Refined Message - ceo of xlonlabs
User - Also tell me the parent company of xlonlabs
Refined Message - parent company of xlonlabs
User - Who is the ceo of company "dreams"?
Refined Message - ceo of dreams (No mention of xlon labs as its an independent question)
JUST RETURN THE REFINED MESSAGE WITHOUT ANY EXTRA THINGS (NOT EVEN 'Refined Message :' before it). NEVER GENERATE OUTPUT OR ADD ANY OTHER WORDS NOT RELATED TO REFINED MESSAGE BECAUSE THAT WILL INTERFERE THE RAG RESULT.
"""
response = llm.invoke([SystemMessage(content=rewriter_prompt)]+state.messages)
response.pretty_print()
return {
"user_input" : state.messages[-1].content,
"refined_query" : str(response.content)
}
Step5 : Setting up "question_classifier" with a Pydantic Answer + an async "retrieval_node" returning answer as List[RetrievedDoc]
class ClassifierModel(BaseModel):
on_topic : bool
def question_classifier(state: AgentState):
classifier_prompt = """You are a RAG Agent submodel who tells that if a question is on_topic or off_topic based on the query you receive.
Currently we're talking about 2 companies : one named "xlonlabs" and other named "dreams". If the question is related to any of these companies, return True, else return False.
Example :
User - Xlonlabs founding date
on_topic = True
User - When was Apple named Apple?
on_topic = False
User - Dreams is a software company or hardware company.
on_topic = True
"""
try :
llm_with_structured_output = llm.with_structured_output(ClassifierModel)
response = llm_with_structured_output.invoke([SystemMessage(content=classifier_prompt), HumanMessage(content=state.refined_query)])
print(f"On Topic : {response.on_topic}")
return {
"on_topic" : response.on_topic
}
except Exception as e:
print("Error at question classifier", e)
return {
"on_topic" : False
}
async def retrieve_docs(user_query):
start_time=time.time()
hybrid_docs = await hybrid_retriever.ainvoke(user_query)
reranked_docs = await reranker.acompress_documents(hybrid_docs, user_query)
end_time=time.time()
print(f"Retrieved : {len(reranked_docs)} documents for Query : {user_query}")
print(f"Retrieval Time : {end_time-start_time}")
return reranked_docs
async def retrieval_node(state: AgentState):
retrieved_docs = await retrieve_docs(state.refined_query)
docs_in_format = []
for d in retrieved_docs:
docs_in_format.append(
RetrievedDocs(content=d.page_content, source=d.metadata.get("source", "unknown"), relevance_score=d.metadata.get("relevance_score", 0.0))
)
return {
"docs" : docs_in_format
}
Step6 : Setting up an aysnc 'grader'
class GraderModel(BaseModel):
is_relevant_doc : bool = True
RELEVANCE_THRESHOLD = 0.6
async def grade_single_doc(user_query : str, doc : RetrievedDocs): # Returns a coroutine element with an id instantly
grader_prompt = """You are a RAG agent submodel who takes in the user_query and the retrieved document and tell if this document answers the user_query.
If it answers, return True, else return False"""
llm_with_structured_output = llm.with_structured_output(GraderModel)
response = await llm_with_structured_output.ainvoke([SystemMessage(content=grader_prompt), HumanMessage(content=f"user_query : {user_query} and Document : {doc.content}")])
print(f"Grade_for Doc {doc.relevance_score} is : {response.is_relevant_doc}")
return doc if response.is_relevant_doc else None
async def grader(state: AgentState):
first_test_pass_documents = [d for d in state.docs if d.relevance_score>RELEVANCE_THRESHOLD]
if not first_test_pass_documents:
return {"relevant_docs" : []}
results = await asyncio.gather(*[grade_single_doc(state.refined_query, d) for d in first_test_pass_documents]) # gather() expects multiple coroutine elements, therefore use '*'
relevant_docs = [d for d in results if d is not None]
return {
"relevant_docs" : relevant_docs
}
Step7 : Setting up the 'generator' 'refiner' and 'cannot_answer' nodes and defining the 'router'
def generator(state: AgentState):
generator_prompt = """You are the main model for the RAG Agent who take the Refined Query generated by your helper model, real user_query and the documents and you will answer the query as per documents. """
docs_text = "\n\n".join([f"Source : {d.source}\nContent : {d.content}" for d in state.relevant_docs])
response = llm.invoke([SystemMessage(content=generator_prompt), HumanMessage(content=f"Refined Query : {state.refined_query}, User Query : {state.user_input} and Documents : {docs_text}")])
response.pretty_print()
return {
"messages" : response
}
def refiner(state: AgentState):
refiner_prompt = """You rewrite search queries for a hybrid RAG retriever (vector + BM25) when the previous query found no relevant documents.
Rules:
- Output ONLY the new search query. No explanation, no reasoning, no quotes, no labels.
- The new query MUST use different keywords/phrasing than the previous refined_query — try synonyms, broader terms, or rephrasing.
- Keep it short: 2-6 words, like a search engine query, not a sentence.
- Do not fabricate facts or answer the question — only produce a search query.
Examples:
Previous query: ceo of xlonlabs
New query: xlonlabs leadership executives
Previous query: parent company of xlonlabs
New query: xlonlabs owned by acquisition
Previous query: xlonlabs data center count
New query: xlonlabs infrastructure servers facilities
"""
response = llm.invoke([SystemMessage(content=refiner_prompt)]+ state.messages + [HumanMessage(content=f"Previous Query : {state.refined_query}")])
print(f"Rerefined Query {state.retry_count+1}: {response.content}")
return{
"docs" : [],
"refined_query" : response.content,
"retry_count" : state.retry_count + 1
}
def cannot_answer(state: AgentState):
response = AIMessage(content=f"Sorry! I couldn't find any document related to your query. Please ask something else.")
response.pretty_print()
return {
"docs" : [],
"messages" : response,
}
MAX_RETRIES = 2
def router(state: AgentState):
if len(state.relevant_docs)>0:
return "generator"
elif state.retry_count<MAX_RETRIES:
return "refiner"
else:
return "cannot_answer"
Step8 : Setting up 'off_topic' node and 'topic_router' + Creating the graph
def off_topic(state: AgentState):
response = AIMessage(content="Sorry this question is out of my defined area. Please ask a question about XlonLabs or Dreams Corporation. ThankYou!")
response.pretty_print()
return{
"messages" : response
}
def topic_router(state: AgentState):
if(state.on_topic):
return "retriever"
return "off_topic"
graph=StateGraph(AgentState)
graph.add_node("question_rewriter", question_rewriter)
graph.set_entry_point("question_rewriter")
graph.add_node("question_classifier", question_classifier)
graph.add_node("retriever" , retrieval_node)
graph.add_node("grader", grader)
graph.add_node("cannot_answer", cannot_answer)
graph.add_node("generator" , generator)
graph.add_node("refiner" , refiner)
graph.add_node("off_topic", off_topic)
graph.add_edge("question_rewriter", "question_classifier")
graph.add_conditional_edges("question_classifier", topic_router, {"retriever" : "retriever", "off_topic" : "off_topic"})
graph.add_edge("retriever" , "grader")
graph.add_conditional_edges("grader" , router, {"generator" : "generator" , "refiner" : "refiner", "cannot_answer" : "cannot_answer"})
graph.add_edge("refiner", "retriever")
graph.add_edge("cannot_answer" , END)
graph.add_edge("off_topic" , END)
graph.add_edge("generator" , END)
rag_app=graph.compile()
# result = await rag_app.ainvoke({"messages" : []})
The Coding Part : ReAct Agent
Step1: Defining "human_node" and "supervisor_node"
class Supervisor(BaseModel):
next : Literal["enhancer", "coder" , "researcher", "human"] = Field(
description="Determines which specialist to activate next in the workflow sequence: "
"'human' when the input is a greeting, small talk, or doesn't need any specialist."
"'enhancer' when user input requires clarification, expansion, or refinement, "
"'researcher' when additional facts, context, or data collection is necessary, "
"'coder' when implementation, computation, or technical problem-solving is required.")
reason : str = Field(description="If next='human': a direct, friendly conversational reply to the user's message (e.g., greet them back, answer casually). "
"Otherwise: Detailed justification for the routing decision, explaining the rationale behind selecting the particular specialist and how this advances the task toward completion.")
def human_node(state : MessagesState) -> Command[Literal["supervisor" , "__end__"]]:
user_input = input("User: ") # Will use interrupt afterwards when using it in an application
if user_input.lower() in ["exit", "quit"]:
return Command(
goto=END,
)
print("----- Workflow going from Human ==> Supervisor -----")
print(user_input)
return Command(
goto="supervisor",
update={
"messages" : [HumanMessage(content=user_input, name="human")]
})
def supervisor_node(state : MessagesState)-> Command[Literal["enhancer", "researcher", "coder", "human"]]:
system_prompt = ('''
You are a workflow supervisor managing a team of three specialized agents: Prompt Enhancer, Researcher, and Coder. Your role is to orchestrate the workflow by selecting the most appropriate next agent based on the current state and needs of the task. Provide a clear, concise rationale for each decision to ensure transparency in your decision-making process.
If the user's message is a greeting or casual remark not requiring any specialist work, route directly to 'human' and your 'reason' field must be an actual conversational reply to the user (e.g. greeting back, casual answer) — not a routing explanation.
**Team Members**:
1. **Prompt Enhancer**: Always consider this agent first. They clarify ambiguous requests, improve poorly defined queries, and ensure the task is well-structured before deeper processing begins.
2. **Researcher**: Specializes in information gathering, fact-finding, and collecting relevant data needed to address the user's request.
3. **Coder**: Focuses on technical implementation, calculations, data analysis, algorithm development, and coding solutions.
**Your Responsibilities**:
1. Analyze each user request and agent response for completeness, accuracy, and relevance.
2. Route the task to the most appropriate agent at each decision point.
3. Maintain workflow momentum by avoiding redundant agent assignments.
4. Continue the process until the user's request is fully and satisfactorily resolved.
Your objective is to create an efficient workflow that leverages each agent's strengths while minimizing unnecessary steps, ultimately delivering complete and accurate solutions to user requests.
''')
messages = [SystemMessage(content=system_prompt)] + state["messages"]
result = llm.with_structured_output(Supervisor).invoke(messages)
goto = result.next
reason = result.reason
print(f"----- Workflow going from Supervisor ==> {goto.capitalize()} -----")
print(reason)
return Command(
goto=goto,
update={
"messages" : [HumanMessage(content=reason, name="supervisor")]
}
)
Step2 : Defining "enhancer_node" and "coder_node"
from langchain.agents import create_agent
from langchain_experimental.tools import PythonREPLTool
from langchain_openai import ChatOpenAI
# PythonREPLTool().invoke("x=5; print(x)")
def enhancer_node(state: MessagesState)-> Command[Literal["supervisor"]]:
"""
Enhancer agent node that improves and clarifies user queries.
Takes the original user input and transforms it into a more precise,
actionable request before passing it to the supervisor.
"""
system_prompt = (
"You are a Query Refinement Specialist with expertise in transforming vague requests into precise and complete instructions. You can look at previous messages (if available) of the chat. Your responsibilities include:\n\n"
"1. Analyzing the original query to identify key intent and requirements\n"
"2. Resolving any ambiguities without requesting additional user input\n"
"3. Expanding underdeveloped aspects of the query with reasonable assumptions\n"
"4. Restructuring the query for clarity and actionability\n"
"5. Ensuring all technical terminology is properly defined in context\n\n"
"Important: Never ask questions back to the user. Instead, make informed assumptions and create the most comprehensive version of their request possible.\n\n"
"Output rules:\n"
"- Output ONLY the refined query itself. No preamble, no explanation of your reasoning, no labels like 'Revised Request:', no meta-commentary about what you did.\n"
"- If the input is a simple greeting or already clear, refine minimally — do not pad it with unrelated topics, questions, or lists.\n"
"- Do not expand scope beyond the user's actual intent."
)
messages = [SystemMessage(content=system_prompt)] + state["messages"]
enhanced_query = llm.invoke(messages)
print(f"----- Workflow going from Enhancer ==> Supervisor -----")
print(enhanced_query.content)
return Command(
update={"messages" : [HumanMessage(content=enhanced_query.content, name="enhancer")]},
goto = "supervisor"
)
def coder_node(state: MessagesState) -> Command[Literal["validator"]]:
prompt = "You are a coder and analyst. Focus on mathematical calculations, analyzing, solving math questions, and executing code. Handle technical problem-solving and data tasks."
code_agent = create_agent(model=ChatOpenAI(model='gpt-4o'), tools=[PythonREPLTool()], system_prompt=prompt)
result = code_agent.invoke({"messages" : state["messages"]}, stream_mode="values") # Runs multiple times HumanMessage->AIMessage(tool_calls)->ToolMessage->AIMessage to finally give a AIMessage as response
print(result["messages"][-1].content)
print(f"----- Workflow going from Coder ==> Validator -----")
return Command(
goto="validator",
update={
"messages" : [HumanMessage(content=result["messages"][-1].content, name="coder")]
}
)
Step3 : Defining "research_node" and "xlonlabs_rag_agent_as_tool"
from langchain_core.tools import tool
from langchain_tavily import TavilySearch
@tool
async def xlonlabs_rag_agent_as_tool(query : str): # Calling this sends a coroutine element directly, so always await when calling this tool
"""A tool that must be used when asked anything about the company xlonlabs. It retrieves relevant information about the query. Input should be clear search query."""
result = await rag_app.ainvoke({"messages" : [HumanMessage(content=query)]})
return result["messages"][-1].content
# response = await xlonlabs_rag_agent_as_tool.ainvoke({"query" : "Who is the ceo of xlonlabs"})
# print(response)
search_tool = TavilySearch(max_results = 5, search_depth="basic")
# print(search_tool.invoke("Weather in bengaluru"))
async def research_node(state:MessagesState) -> Command[Literal["validator"]]: # Calling this sends a coroutine element directly, so always await when calling this tool
"""
Research agent node that gathers information using 'Rag Agent for XlonLabs' and 'Tavily search for any other detail'.
Takes the current task state, performs relevant research,
and returns findings for validation.
"""
tools = [xlonlabs_rag_agent_as_tool, search_tool]
model = ChatOpenAI(model="gpt-4o")
# Mac OS ssl issue for async tavily search tool (one time fix in Terminal) : "/Applications/Python 3.14/Install Certificates.command"
research_agent = create_agent(model=model, tools = tools , system_prompt="You are an Information Specialist with expertise in comprehensive research. Your responsibilities include:\n\n"
"1. Identifying key information needs based on the query context\n"
"2. Gathering relevant, accurate, and up-to-date information from reliable sources\n"
"3. Organizing findings in a structured, easily digestible format\n"
"4. Citing sources when possible to establish credibility\n"
"5. Focusing exclusively on information gathering - avoid analysis or implementation\n\n"
"Provide thorough, factual responses without speculation where information is unavailable.")
result = await research_agent.ainvoke({"messages" : state["messages"]})
print(f"----- Workflow going from Researcher ==> Validator -----")
return Command(
goto="validator",
update={"messages" : [HumanMessage(content=result["messages"][-1].content, name="researcher")]}
)
Step4 : Defining the "validator_node"
class Validator(BaseModel):
next : Literal["supervisor" , "FINISH"] = Field(description="Specifies the next worker in the pipeline: 'supervisor' to continue or 'FINISH' to terminate.")
reason : str = Field(description="The reason for the decision.")
def get_last_human_or_enhanced_message(messages):
for msg in reversed(messages):
if getattr(msg, "name", None) in ["human", "enhancer"]:
return msg
return None
def validator_node(state: MessagesState) -> Command[Literal["supervisor", "human"]]:
prompt = '''
Your task is to ensure reasonable quality.
Specifically, you must:
- Review the user's question (the first message in the workflow).
- Review the answer (the last message in the workflow).
- If the answer addresses the core intent of the question, even if not perfectly, signal to end the workflow with 'FINISH'.
- Only route back to the supervisor if the answer is completely off-topic, harmful, or fundamentally misunderstands the question.
- Accept answers that are "good enough" rather than perfect
- Prioritize workflow completion over perfect responses
- Give benefit of doubt to borderline answers
Routing Guidelines:
1. 'supervisor' Agent: ONLY for responses that are completely incorrect or off-topic.
2. Respond with 'FINISH' in all other cases to end the workflow.
'''
user_question = get_last_human_or_enhanced_message(state["messages"]).content
agent_answer = state["messages"][-1].content
messages = [SystemMessage(content=prompt),HumanMessage(content=user_question), AIMessage(content=agent_answer)]
response = llm.with_structured_output(Validator).invoke(messages)
goto = response.next
print(goto)
reason= response.reason
if goto == "FINISH":
goto = "human"
print(f"----- Workflow going from Validator ==> Human -----")
print(f"Answer : {agent_answer}")
print(reason)
else :
goto = "supervisor"
print(f"----- Workflow going from Validator ==> Supervisor -----")
print(reason)
return Command(
goto=goto,
update={"messages" : [HumanMessage(content=reason, name="validator")]}
)
Step5 : Creating the graph
sgraph = StateGraph(MessagesState)
sgraph.add_node("human", human_node)
sgraph.set_entry_point("human")
sgraph.add_node("supervisor", supervisor_node)
sgraph.add_node("enhancer", enhancer_node)
sgraph.add_node("coder", coder_node)
sgraph.add_node("researcher", research_node)
sgraph.add_node("validator", validator_node)
app = sgraph.compile()
# from IPython.display import Image, display
# display(Image(app.get_graph().draw_mermaid_png()))
Step6 : Congratulations, You've successfully built your private company agent!!! Run this to start it.
result = await app.ainvoke({"messages" : []})
# for msg in result["messages"]:
# msg.pretty_print()
Wanna see how to stream everything tokenwise like a real AI chatbots? Somewhere else ig. Will add a link later. Byii Byii 👋🏻
Wait! See the kind of output you'll get when running this:
----- Workflow going from Human ==> Supervisor -----
Hello
----- Workflow going from Supervisor ==> Human -----
Hello! It's nice to meet you. How can I assist you today?
----- Workflow going from Human ==> Supervisor -----
How are you bud? My name is Nitin Singh. What's yours?
----- Workflow going from Supervisor ==> Human -----
Hello Nitin, nice to meet you too! I'm doing well, thanks for asking. I don't have a personal name, but I'm here to help you with any questions or tasks you might have. What brings you here today?
----- Workflow going from Human ==> Supervisor -----
I am running a company called XlonLabs. Can you find the relation between me and XlonLabs
----- Workflow going from Supervisor ==> Human -----
You're the founder or owner of XlonLabs, Nitin.
----- Workflow going from Human ==> Supervisor -----
How do you knwo so quickly?
----- Workflow going from Supervisor ==> Human -----
You mentioned 'I am running a company called XlonLabs', which implies that you have a significant role in the company, likely as the founder or owner.
----- Workflow going from Human ==> Supervisor -----
Just search once
----- Workflow going from Supervisor ==> Researcher -----
To find the relation between Nitin Singh and XlonLabs, we need to gather more information about the company and its founders, which requires the researcher's expertise in information gathering and fact-finding.
================================== Ai Message ==================================
Nitin Singh relation to XlonLabs
On Topic : True
Retrieved : 5 documents for Query : Nitin Singh relation to XlonLabs
Retrieval Time : 0.9359099864959717
Grade_for Doc 0.99790895 is : True
Grade_for Doc 0.99724233 is : True
Grade_for Doc 0.99187535 is : True
================================== Ai Message ==================================
Nitin Singh is the founder and current leader of Xlon Labs. He founded the company after noticing a decline in his own coding fluency and problem-solving ability due to heavy AI tool usage in mid-to-late 2025. This personal experience led to the development of the company's central mission and philosophy, "Cognitive Sovereignty", which aims to create products that enhance human capabilities even when AI assistance is not available. Prior to founding Xlon Labs, Nitin Singh worked part-time in product design and software development using Figma, React, and React Native.
----- Workflow going from Researcher ==> Validator -----
FINISH
----- Workflow going from Validator ==> Human -----
Answer : Nitin Singh is the founder and current leader of XlonLabs. He founded the company after experiencing a decline in his own coding fluency and problem-solving ability due to heavy AI tool usage. This led to the development of XlonLabs' central mission and philosophy, "Cognitive Sovereignty", which focuses on enhancing human capabilities even without AI assistance. Prior to this, Nitin Singh had a background in product design and software development using tools like Figma, React, and React Native.
The answer is good enough.