Build a Multi-Agent Operations Copilot
Orchestrate four LangGraph agents over industrial SOPs to generate structured incident reports.
Introduction
30 Second Summary
When something goes wrong on a factory floor, operators need answers fast. They shouldn't have to dig through hundreds of pages of safety manuals and maintenance procedures to figure out what's happening and what to do next.
In this project, you will build an Industrial Operations Copilot that uses multiple AI agents to analyze operator queries, retrieve relevant procedures, and generate structured incident reports automatically.
What You'll Build
A multi-agent pipeline built with LangGraph that orchestrates four specialized AI agents to transform plain-language operator queries into structured incident reports with confidence scores and cited sources.
By the end of this project, you'll have:
- A document ingestion pipeline that chunks and embeds industrial SOP documents into a ChromaDB vector store.
- A LangGraph StateGraph with four agent nodes that analyze queries, retrieve documents, generate reports, and validate outputs.
- Structured Pydantic incident report models with severity ratings, root causes, recommended actions, and confidence scores.
- Secret Mission: Add a metrics tracking system that logs agent latencies, token usage, and confidence score distributions across queries.
Are there any prerequisites?
You should be comfortable with basic Python (functions, dictionaries, imports). Familiarity with RAG concepts like embeddings and vector stores is helpful but not required.
You'll also need a free Google AI API key. The setup step walks you through getting one from Google AI Studio.
Define Pydantic Models and LangGraph State
Your environment is ready and your API key works. Now you need to define the data structures that flow between your agents.
Each agent in your pipeline produces structured output. Without a shared schema, agents would pass unstructured text to each other, making validation and routing impossible. In this step, you will create Pydantic models that define exactly what each agent outputs, and a LangGraph state schema that carries data through the entire pipeline.
In this step, get ready to:
- Create Pydantic models for query analysis, incident reports, and validation results.
- Define the LangGraph TypedDict state schema that flows through all agents.
- Verify that your models work with Gemini's structured output.
Create the Pydantic output models
Gemini can return structured JSON instead of free-form text when you give it a Pydantic model as a schema. Your pipeline uses three models: one for analyzing queries, one for incident reports, and one for validation checks.
- Create a new file called models.py in your project directory.
- Add the imports and the first model, QueryAnalysis, by pasting this code:
"""Pydantic models for structured agent outputs."""
from datetime import datetime
from pydantic import BaseModel, Field
class QueryAnalysis(BaseModel):
"""Structured analysis of an operator query."""
intent: str = Field(description="The intent of the query: troubleshoot, maintenance, or status_check")
equipment_id: str = Field(description="The equipment identifier mentioned in the query")
symptoms: list[str] = Field(description="List of symptoms or issues described")
urgency: str = Field(description="Urgency level: low, medium, high, or critical")What does this code do?
Each field uses Pydantic's Field with a description parameter. Gemini reads these descriptions to understand what value to put in each field.
When the Query Analyzer agent runs, it will return an instance of QueryAnalysis with the equipment ID, symptoms, intent, and urgency extracted from the operator's question.
- Below QueryAnalysis, add the IncidentReport model:
class IncidentReport(BaseModel):
"""Structured incident report generated from SOP analysis."""
severity: str = Field(description="Severity level: low, medium, high, or critical")
root_cause: str = Field(description="Most likely root cause based on SOP analysis")
recommended_actions: list[str] = Field(description="Ordered list of recommended actions from SOPs")
confidence_score: float = Field(description="Confidence score between 0.0 and 1.0")
cited_sources: list[str] = Field(description="List of SOP document titles used as sources")
timestamp: str = Field(default_factory=lambda: datetime.now().isoformat())What does this model do?
This is the main output of your pipeline. The Generator agent produces an IncidentReport with a root cause analysis, recommended actions from SOPs, and a confidence score.
The confidence_score field is what the Validator agent checks. If the score is too low, the pipeline loops back for another attempt. The cited_sources field ensures traceability back to the original SOP documents.
- Below IncidentReport, add the final model, ValidationResult:
class ValidationResult(BaseModel):
"""Result of the validator agent's assessment."""
is_valid: bool = Field(description="Whether the report meets quality thresholds")
issues: list[str] = Field(description="List of issues found, empty if valid")What does this model do?
The Validator agent uses this model to report whether the incident report passes quality checks. The issues list captures specific problems like low confidence or missing citations.
- Save models.py.
Seeing a syntax error?
Make sure each class is separated by two blank lines (Python convention). Check that the list[str] type hints use lowercase list, not List. This syntax requires Python 3.10+.
Help me fix a syntax error in my Pydantic models
✔️ Awesome, I've got everything!
Great. Double check you have saved models.py before moving on.
ⓧ I'd like to double check the full code
"""Pydantic models for structured agent outputs."""
from datetime import datetime
from pydantic import BaseModel, Field
class QueryAnalysis(BaseModel):
"""Structured analysis of an operator query."""
intent: str = Field(description="The intent of the query: troubleshoot, maintenance, or status_check")
equipment_id: str = Field(description="The equipment identifier mentioned in the query")
symptoms: list[str] = Field(description="List of symptoms or issues described")
urgency: str = Field(description="Urgency level: low, medium, high, or critical")
class IncidentReport(BaseModel):
"""Structured incident report generated from SOP analysis."""
severity: str = Field(description="Severity level: low, medium, high, or critical")
root_cause: str = Field(description="Most likely root cause based on SOP analysis")
recommended_actions: list[str] = Field(description="Ordered list of recommended actions from SOPs")
confidence_score: float = Field(description="Confidence score between 0.0 and 1.0")
cited_sources: list[str] = Field(description="List of SOP document titles used as sources")
timestamp: str = Field(default_factory=lambda: datetime.now().isoformat())
class ValidationResult(BaseModel):
"""Result of the validator agent's assessment."""
is_valid: bool = Field(description="Whether the report meets quality thresholds")
issues: list[str] = Field(description="List of issues found, empty if valid")Define the LangGraph state schema
LangGraph uses a typed dictionary to carry data between nodes in your graph. Every agent reads from and writes to this shared state object. The state schema defines which fields exist and what types they hold.
- Create a new file called state.py in your project directory.
- Add the pipeline state definition by pasting this code:
"""LangGraph state definition for the multi-agent pipeline."""
from typing import TypedDict
class PipelineState(TypedDict):
query: str
analysis: dict | None
retrieved_docs: list[dict]
report: dict | None
validation_passed: bool
retry_count: intWhat does this state carry?
- query holds the original operator question.
- analysis stores the Query Analyzer's structured output (equipment ID, symptoms, urgency).
- retrieved_docs holds the SOP documents pulled from ChromaDB.
- report stores the generated incident report.
- validation_passed and retry_count control the validation loop. If validation fails, the pipeline retries generation.
- Save state.py.
- Confirm both files exist by running:
ls models.py state.pyYou should see both filenames printed without any errors.
File not found?
Make sure your terminal is in the same directory where you created the files. Use ls to list files in the current directory and confirm your location.
Help me find my project files
✔️ Awesome, I've got everything!
Great. Double check you have saved state.py before moving on.
ⓧ I'd like to double check the full code
"""LangGraph state definition for the multi-agent pipeline."""
from typing import TypedDict
class PipelineState(TypedDict):
query: str
analysis: dict | None
retrieved_docs: list[dict]
report: dict | None
validation_passed: bool
retry_count: intTest structured output with Gemini
Before building the full pipeline, confirm that Gemini can actually return data in your Pydantic schema format. The with_structured_output method binds a Pydantic model to the LLM so every response is parsed into that model automatically.
- Run this quick test in your terminal to verify that your QueryAnalysis model works with Gemini:
python3 -c "
from langchain_google_genai import ChatGoogleGenerativeAI
from models import QueryAnalysis
import os
llm = ChatGoogleGenerativeAI(model='gemini-3.5-flash', google_api_key=os.environ.get('GOOGLE_API_KEY'), temperature=0.2)
structured_llm = llm.with_structured_output(QueryAnalysis)
result = structured_llm.invoke('Pump P-401 showing high vibration and elevated temperature')
print(result)
"What does this test do?
It creates a ChatGoogleGenerativeAI instance and binds your QueryAnalysis model to it using with_structured_output. When you invoke the LLM with an operator query, Gemini returns a JSON object that Pydantic validates against your schema.
The default method is json_schema, which uses Gemini's native structured output API for reliable parsing.
You should see a QueryAnalysis object printed with fields like intent, equipment_id, symptoms, and urgency filled in by Gemini.
Getting an API error?
Make sure your GOOGLE_API_KEY environment variable is still set in this terminal session. If you opened a new terminal, you need to export it again.
Also confirm you are running the command from the same directory that contains models.py. Python needs to import your file directly.
Help me debug my structured output test
Why this approach over raw JSON parsing?
You could ask Gemini for JSON and parse it yourself. But with_structured_output handles schema enforcement, retry on malformed responses, and type validation in one call. Your agent code stays clean because it always receives a typed Python object, never a raw string.
Your data contracts are locked in. Every agent in the pipeline now knows exactly what shape of data it receives and produces. Next up, you will build the document ingestion pipeline that fills ChromaDB with SOP documents for your agents to search.
Build the RAG Document Ingestion Pipeline
Your Pydantic models define the shape of data flowing through the pipeline. Your state schema connects every agent. Now you need the knowledge base that your agents will reason over.
In a RAG system, the quality of retrieved documents directly determines the quality of generated outputs. In this step, you will build the document ingestion layer that chunks industrial SOPs, embeds them into ChromaDB, and retrieves the most relevant procedures for any operator query.
In this step, get ready to:
- Create sample industrial SOP documents as structured Python data.
- Ingest documents into ChromaDB with metadata filtering.
- Implement a retrieval function that returns ranked document chunks.
Create the SOP document corpus
Your copilot needs a knowledge base of Standard Operating Procedures to reason over. Real industrial environments store these as PDFs or wiki pages. For this project, you will define four representative SOPs as Python strings with rich metadata.
- Create a new file called ingest.py in your project directory.
- Add the following document corpus by pasting this code into ingest.py:
"""Document ingestion pipeline for industrial SOPs."""
import chromadb
SOP_DOCUMENTS = [
{
"title": "SOP-PUMP-401: Centrifugal Pump Maintenance",
"content": (
"Pump P-401 is a centrifugal pump used in the cooling water circuit. "
"High vibration readings above 7.1 mm/s indicate bearing wear or misalignment. "
"Recommended actions: 1) Check bearing condition using vibration spectrum analysis. "
"2) Verify shaft alignment with laser alignment tool. "
"3) Inspect coupling for wear or damage. "
"4) If vibration exceeds 11.2 mm/s, shut down immediately and replace bearings. "
"Elevated temperature above 85C at the bearing housing suggests inadequate lubrication "
"or bearing failure. Combine with vibration data for root cause determination."
),
"metadata": {"equipment_type": "pump", "procedure_category": "maintenance", "severity_level": "high"},
},
{
"title": "SOP-TEMP-200: Temperature Monitoring Procedures",
"content": (
"All rotating equipment must have temperature monitored at bearing housings. "
"Normal operating range for P-series pumps is 40-75C. "
"Warning threshold: 75-85C requires increased monitoring frequency to every 2 hours. "
"Critical threshold: above 85C requires immediate investigation. "
"Common causes of elevated temperature: insufficient lubrication, "
"contaminated lubricant, excessive load, or bearing degradation. "
"Cross-reference with vibration data from SOP-VIB-100 for comprehensive diagnosis."
),
"metadata": {"equipment_type": "pump", "procedure_category": "monitoring", "severity_level": "medium"},
},
{
"title": "SOP-VIB-100: Vibration Analysis Standard",
"content": (
"Vibration analysis is the primary predictive maintenance tool for rotating equipment. "
"ISO 10816 velocity standards for Group 2 machines: "
"Good: below 2.8 mm/s. Satisfactory: 2.8-7.1 mm/s. "
"Unsatisfactory: 7.1-11.2 mm/s. Unacceptable: above 11.2 mm/s. "
"When high vibration coincides with elevated temperature, "
"the most likely root cause is bearing failure. "
"Priority actions: 1) Collect vibration spectrum. 2) Check for 1x and 2x RPM peaks "
"indicating imbalance or misalignment. 3) Schedule bearing replacement within 72 hours "
"if trending upward. 4) For combined high vibration and high temperature, "
"reduce load by 50% until maintenance can be performed."
),
"metadata": {"equipment_type": "pump", "procedure_category": "predictive_maintenance", "severity_level": "high"},
},
{
"title": "SOP-EMER-050: Emergency Shutdown Procedures",
"content": (
"Emergency shutdown is required when any of the following conditions are met: "
"1) Vibration exceeds 11.2 mm/s on any rotating equipment. "
"2) Bearing temperature exceeds 100C. "
"3) Unusual noise indicating catastrophic failure. "
"4) Visible smoke or sparks from equipment. "
"Shutdown procedure: Activate local E-stop, notify control room, "
"isolate energy sources (electrical, hydraulic, pneumatic), "
"apply lockout/tagout before any inspection. "
"Post-shutdown: Complete incident report within 4 hours."
),
"metadata": {"equipment_type": "general", "procedure_category": "emergency", "severity_level": "critical"},
},
]What does this code do?
Each SOP document has three fields that mirror how real industrial document management systems work:
- title identifies the procedure (e.g. SOP-PUMP-401) so your agents can cite it in reports.
- content holds the actual procedure text that gets embedded and retrieved.
- metadata tags each document with equipment_type, procedure_category, and severity_level for filtered retrieval.
You won't see a change yet. The next chunk uses these documents to populate ChromaDB.
Build the ingestion function
Raw documents need to be embedded into a vector store before your agents can search them. ChromaDB handles embedding automatically using its default model. You just need to feed it the documents with their metadata.
- Add the create_vector_store() function to the bottom of ingest.py:
def create_vector_store() -> chromadb.Collection:
"""Create and populate the ChromaDB collection with SOP documents."""
# Initialize a persistent ChromaDB client stored in ./chroma_data
client = chromadb.PersistentClient(path="./chroma_data")
collection = client.get_or_create_collection(
name="industrial_sops",
metadata={"description": "Industrial Standard Operating Procedures"},
)
# Skip ingestion if documents are already loaded
existing = collection.count()
if existing > 0:
return collection
# Prepare documents, metadata, and unique IDs for batch insertion
documents = []
metadatas = []
ids = []
for i, doc in enumerate(SOP_DOCUMENTS):
documents.append(doc["content"])
metadatas.append({**doc["metadata"], "title": doc["title"]})
ids.append(f"sop_{i}")
# Add all documents to the collection in one batch
collection.add(documents=documents, metadatas=metadatas, ids=ids)
return collectionWhat does this code do?
- PersistentClient(path="./chroma_data") stores the vector database on disk so embeddings survive between script runs.
- get_or_create_collection ensures you can call this function multiple times without duplicating data.
- The if existing > 0 check acts as an idempotency guard. If documents are already ingested, it returns immediately.
- collection.add() embeds the document text using ChromaDB's default embedding model and stores it alongside the metadata.
- Save ingest.py.
- Test the ingestion by running this command:
python -c "from ingest import create_vector_store; c = create_vector_store(); print(f'Collection has {c.count()} documents')"You should see Collection has 4 documents printed in your terminal.
Seeing a ModuleNotFoundError?
- Make sure you installed chromadb in Step 1. Run pip install chromadb if it is missing.
- Confirm you are running the command from your project directory (the folder containing ingest.py).
Help me debug a ModuleNotFoundError when importing chromadb in my ingest.py file.
Implement the retrieval function
Your agents need a way to search the vector store and get back ranked results. The retrieval function takes a natural-language query, finds the closest document embeddings, and returns them with distance scores so downstream agents can judge relevance.
- Add the retrieve_documents() function to the bottom of ingest.py:
def retrieve_documents(collection: chromadb.Collection, query: str, n_results: int = 3) -> list[dict]:
"""Retrieve relevant SOP documents from ChromaDB."""
# Query the collection using semantic similarity search
results = collection.query(query_texts=[query], n_results=n_results)
# Reshape results into a list of document dicts with metadata and distance
retrieved = []
for i in range(len(results["ids"][0])):
retrieved.append({
"id": results["ids"][0][i],
"content": results["documents"][0][i],
"metadata": results["metadatas"][0][i],
"distance": results["distances"][0][i],
})
return retrievedWhat does this code do?
- collection.query(query_texts=[query]) embeds the query string and finds the n_results nearest documents by cosine similarity.
- The function reshapes ChromaDB's nested list format into a flat list of dictionaries. Each dict contains the document content, metadata, and a distance score (lower means more relevant).
- This output format matches what the retrieved_docs field in your PipelineState expects.
- Save ingest.py.
- Test the full pipeline by running this command:
python -c "from ingest import create_vector_store, retrieve_documents; c = create_vector_store(); docs = retrieve_documents(c, 'pump vibration high temperature'); print(f'Retrieved {len(docs)} docs'); print(f'Top result: {docs[0][\"metadata\"][\"title\"]}')"
You should see 3 documents retrieved, with the top result being one of the pump-related SOPs.
Getting zero results or an error?
- If you see 0 documents, delete the ./chroma_data folder and run the test again. The idempotency check may have stored an empty collection from a previous failed attempt.
- Check that your SOP_DOCUMENTS list has exactly 4 entries. A missing comma between dictionary entries causes a silent merge.
Help me debug why my ChromaDB retrieval is returning zero results or throwing an error.
✔️ Awesome, I've got everything!
Great. Double check you have saved ingest.py with all three sections: the SOP_DOCUMENTS list, create_vector_store(), and retrieve_documents().
ⓧ I'd like to double check the full code
Here is the complete ingest.py file:
"""Document ingestion pipeline for industrial SOPs."""
import chromadb
SOP_DOCUMENTS = [
{
"title": "SOP-PUMP-401: Centrifugal Pump Maintenance",
"content": (
"Pump P-401 is a centrifugal pump used in the cooling water circuit. "
"High vibration readings above 7.1 mm/s indicate bearing wear or misalignment. "
"Recommended actions: 1) Check bearing condition using vibration spectrum analysis. "
"2) Verify shaft alignment with laser alignment tool. "
"3) Inspect coupling for wear or damage. "
"4) If vibration exceeds 11.2 mm/s, shut down immediately and replace bearings. "
"Elevated temperature above 85C at the bearing housing suggests inadequate lubrication "
"or bearing failure. Combine with vibration data for root cause determination."
),
"metadata": {"equipment_type": "pump", "procedure_category": "maintenance", "severity_level": "high"},
},
{
"title": "SOP-TEMP-200: Temperature Monitoring Procedures",
"content": (
"All rotating equipment must have temperature monitored at bearing housings. "
"Normal operating range for P-series pumps is 40-75C. "
"Warning threshold: 75-85C requires increased monitoring frequency to every 2 hours. "
"Critical threshold: above 85C requires immediate investigation. "
"Common causes of elevated temperature: insufficient lubrication, "
"contaminated lubricant, excessive load, or bearing degradation. "
"Cross-reference with vibration data from SOP-VIB-100 for comprehensive diagnosis."
),
"metadata": {"equipment_type": "pump", "procedure_category": "monitoring", "severity_level": "medium"},
},
{
"title": "SOP-VIB-100: Vibration Analysis Standard",
"content": (
"Vibration analysis is the primary predictive maintenance tool for rotating equipment. "
"ISO 10816 velocity standards for Group 2 machines: "
"Good: below 2.8 mm/s. Satisfactory: 2.8-7.1 mm/s. "
"Unsatisfactory: 7.1-11.2 mm/s. Unacceptable: above 11.2 mm/s. "
"When high vibration coincides with elevated temperature, "
"the most likely root cause is bearing failure. "
"Priority actions: 1) Collect vibration spectrum. 2) Check for 1x and 2x RPM peaks "
"indicating imbalance or misalignment. 3) Schedule bearing replacement within 72 hours "
"if trending upward. 4) For combined high vibration and high temperature, "
"reduce load by 50% until maintenance can be performed."
),
"metadata": {"equipment_type": "pump", "procedure_category": "predictive_maintenance", "severity_level": "high"},
},
{
"title": "SOP-EMER-050: Emergency Shutdown Procedures",
"content": (
"Emergency shutdown is required when any of the following conditions are met: "
"1) Vibration exceeds 11.2 mm/s on any rotating equipment. "
"2) Bearing temperature exceeds 100C. "
"3) Unusual noise indicating catastrophic failure. "
"4) Visible smoke or sparks from equipment. "
"Shutdown procedure: Activate local E-stop, notify control room, "
"isolate energy sources (electrical, hydraulic, pneumatic), "
"apply lockout/tagout before any inspection. "
"Post-shutdown: Complete incident report within 4 hours."
),
"metadata": {"equipment_type": "general", "procedure_category": "emergency", "severity_level": "critical"},
},
]
def create_vector_store() -> chromadb.Collection:
"""Create and populate the ChromaDB collection with SOP documents."""
client = chromadb.PersistentClient(path="./chroma_data")
collection = client.get_or_create_collection(
name="industrial_sops",
metadata={"description": "Industrial Standard Operating Procedures"},
)
existing = collection.count()
if existing > 0:
return collection
documents = []
metadatas = []
ids = []
for i, doc in enumerate(SOP_DOCUMENTS):
documents.append(doc["content"])
metadatas.append({**doc["metadata"], "title": doc["title"]})
ids.append(f"sop_{i}")
collection.add(documents=documents, metadatas=metadatas, ids=ids)
return collection
def retrieve_documents(collection: chromadb.Collection, query: str, n_results: int = 3) -> list[dict]:
"""Retrieve relevant SOP documents from ChromaDB."""
results = collection.query(query_texts=[query], n_results=n_results)
retrieved = []
for i in range(len(results["ids"][0])):
retrieved.append({
"id": results["ids"][0][i],
"content": results["documents"][0][i],
"metadata": results["metadatas"][0][i],
"distance": results["distances"][0][i],
})
return retrievedYour knowledge base is loaded and searchable. Next up, you will build the four agent nodes that use this retrieval layer to analyze queries and generate structured incident reports.
Implement the Four Agent Nodes
Your SOP documents are indexed and ready for retrieval. Now you need the agents that actually process operator queries.
Each agent is a Python function that takes the pipeline state, performs one job, and returns an update to the state. You will build four agents that form a sequential chain: analyze the query, retrieve relevant SOPs, generate a report, then validate it.
In this step, get ready to:
- Build the query analyzer and retriever agent nodes.
- Build the generator agent node that produces structured incident reports.
- Build the validator agent node that enforces quality thresholds.
Build the query analyzer and retriever nodes
The query analyzer is the first agent in the pipeline. It uses Gemini with structured output to extract equipment IDs, symptoms, and urgency from the raw operator text.
- Create a new file called agents.py in your project folder.
- Add the imports, LLM configuration, and vector store initialization by pasting this code:
"""Agent node functions for the LangGraph pipeline."""
import os
from langchain_google_genai import ChatGoogleGenerativeAI
from state import PipelineState
from models import QueryAnalysis, IncidentReport, ValidationResult
from ingest import create_vector_store, retrieve_documents
# Initialize the Gemini LLM with low temperature for consistent outputs
llm = ChatGoogleGenerativeAI(
model="gemini-3.5-flash",
google_api_key=os.environ.get("GOOGLE_API_KEY"),
temperature=0.2,
)
# Load the ChromaDB collection at module level
collection = create_vector_store()What does this code do?
The ChatGoogleGenerativeAI class from langchain-google-genai is the primary interface for calling Gemini models through LangChain. Setting temperature=0.2 keeps outputs deterministic for operational analysis.
Calling create_vector_store() at module level loads the existing ChromaDB collection you populated in the previous step. All four agent functions share this single collection instance.
- Below the module-level setup, add the query_analyzer_node function:
def query_analyzer_node(state: PipelineState) -> dict:
"""Analyze the operator query to extract structured information."""
# Create a structured output chain bound to QueryAnalysis schema
structured_llm = llm.with_structured_output(QueryAnalysis)
prompt = (
"You are an industrial operations analyst. Analyze this operator query "
"and extract the equipment ID, symptoms, intent, and urgency level.\n\n"
f"Query: {state['query']}"
)
# Invoke returns a QueryAnalysis instance directly
analysis = structured_llm.invoke(prompt)
return {"analysis": analysis.model_dump()}What does with_structured_output do?
The with_structured_output(QueryAnalysis) method wraps the LLM so that its response is automatically parsed into your Pydantic model. Instead of raw text, structured_llm.invoke() returns a QueryAnalysis instance with typed fields like equipment_id, symptoms, and urgency.
Calling analysis.model_dump() converts the Pydantic object to a plain dictionary so it can flow through the LangGraph TypedDict state without serialization issues.
- Next, add the retriever_node function below query_analyzer_node:
def retriever_node(state: PipelineState) -> dict:
"""Retrieve relevant SOP documents based on the analyzed query."""
analysis = state["analysis"]
# Combine equipment ID, symptoms, and intent into a rich search query
search_query = (
f"{analysis['equipment_id']} {' '.join(analysis['symptoms'])} "
f"{analysis['intent']}"
)
docs = retrieve_documents(collection, search_query, n_results=3)
return {"retrieved_docs": docs}What does this code do?
The retriever builds a search query by combining the structured fields the analyzer extracted. This gives ChromaDB a richer embedding to match against than the raw operator text alone.
It returns the top 3 matching SOP chunks as a list of dictionaries containing the document content, metadata, and similarity distance.
You won't see a change yet. The next function builds on these two.
Build the generator node
The generator is the core reasoning agent. It takes the retrieved SOPs, combines them with the original query and analysis, then produces a full structured incident report with a confidence score.
- Add the generator_node function below retriever_node in agents.py:
def generator_node(state: PipelineState) -> dict:
"""Generate a structured incident report from retrieved SOPs."""
structured_llm = llm.with_structured_output(IncidentReport)
# Build context string from all retrieved documents
context = "\n\n".join(
f"Source: {doc['metadata']['title']}\n{doc['content']}"
for doc in state["retrieved_docs"]
)
prompt = (
"You are an industrial operations copilot. Based on the following SOP documents "
"and the operator's query, generate a structured incident report.\n\n"
"IMPORTANT: Set confidence_score between 0.0 and 1.0 based on how well the "
"retrieved SOPs cover the reported issue. Only cite sources you actually used.\n\n"
f"Operator Query: {state['query']}\n\n"
f"Equipment Analysis: {state['analysis']}\n\n"
f"Retrieved SOP Context:\n{context}"
)
# Returns an IncidentReport instance with severity, root_cause, actions, etc.
report = structured_llm.invoke(prompt)
return {"report": report.model_dump()}What does this code do?
The generator concatenates each retrieved document with its source title so the LLM can attribute its citations correctly. The prompt explicitly instructs Gemini to set a confidence score and only cite documents it actually referenced.
The output is a complete IncidentReport with severity, root cause, recommended actions, confidence score, and cited sources. This structured output feeds directly into the validator.
You won't see a change yet. The validator uses this output next.
Build the validator node
The validator is a rules-based quality gate. It does not call the LLM. Instead, it checks three conditions: confidence score meets the 0.7 threshold, every cited source exists in the retrieved documents, and the report includes recommended actions.
- Add the validator_node function at the bottom of agents.py:
def validator_node(state: PipelineState) -> dict:
"""Validate the generated report for quality and confidence."""
report = state["report"]
retrieved_titles = [doc["metadata"]["title"] for doc in state["retrieved_docs"]]
issues = []
# Check 1: confidence score must meet threshold
if report["confidence_score"] < 0.7:
issues.append(f"Low confidence score: {report['confidence_score']:.2f}")
# Check 2: every cited source must exist in retrieved docs
for source in report["cited_sources"]:
if source not in retrieved_titles:
issues.append(f"Citation not found in retrieved docs: {source}")
# Check 3: report must include recommended actions
if not report["recommended_actions"]:
issues.append("No recommended actions provided")
validation_passed = len(issues) == 0
retry_count = state["retry_count"] + (0 if validation_passed else 1)
return {
"validation_passed": validation_passed,
"retry_count": retry_count,
}Why is the validator rules-based?
Using deterministic checks instead of another LLM call makes validation fast, predictable, and free of hallucination risk. The validator either passes or fails. No ambiguity.
When validation fails, retry_count increments. In the next step you will wire a conditional edge that routes the state back to the generator for a second attempt when validation_passed is False.
- Save agents.py.
- Verify the file imports correctly by running this command:
python -c "from agents import query_analyzer_node, retriever_node, generator_node, validator_node; print('All agents loaded successfully')"You should see All agents loaded successfully printed in the terminal.
Seeing an import error?
- If you see ModuleNotFoundError: No module named 'langchain_google_genai', confirm you installed the package with pip install langchain-google-genai in the same environment you are running Python from.
- If you see ModuleNotFoundError: No module named 'state', make sure you are running the command from inside your project folder where state.py lives.
- Check that your GOOGLE_API_KEY environment variable is still set. The LLM initialization reads it at import time.
Why is my agents.py failing to import?
✔️ Awesome, I've got everything!
Great. Double check you have saved agents.py and the import test passes.
ⓧ I'd like to double check the full code
"""Agent node functions for the LangGraph pipeline."""
import os
from langchain_google_genai import ChatGoogleGenerativeAI
from state import PipelineState
from models import QueryAnalysis, IncidentReport, ValidationResult
from ingest import create_vector_store, retrieve_documents
llm = ChatGoogleGenerativeAI(
model="gemini-3.5-flash",
google_api_key=os.environ.get("GOOGLE_API_KEY"),
temperature=0.2,
)
collection = create_vector_store()
def query_analyzer_node(state: PipelineState) -> dict:
"""Analyze the operator query to extract structured information."""
structured_llm = llm.with_structured_output(QueryAnalysis)
prompt = (
"You are an industrial operations analyst. Analyze this operator query "
"and extract the equipment ID, symptoms, intent, and urgency level.\n\n"
f"Query: {state['query']}"
)
analysis = structured_llm.invoke(prompt)
return {"analysis": analysis.model_dump()}
def retriever_node(state: PipelineState) -> dict:
"""Retrieve relevant SOP documents based on the analyzed query."""
analysis = state["analysis"]
search_query = (
f"{analysis['equipment_id']} {' '.join(analysis['symptoms'])} "
f"{analysis['intent']}"
)
docs = retrieve_documents(collection, search_query, n_results=3)
return {"retrieved_docs": docs}
def generator_node(state: PipelineState) -> dict:
"""Generate a structured incident report from retrieved SOPs."""
structured_llm = llm.with_structured_output(IncidentReport)
context = "\n\n".join(
f"Source: {doc['metadata']['title']}\n{doc['content']}"
for doc in state["retrieved_docs"]
)
prompt = (
"You are an industrial operations copilot. Based on the following SOP documents "
"and the operator's query, generate a structured incident report.\n\n"
"IMPORTANT: Set confidence_score between 0.0 and 1.0 based on how well the "
"retrieved SOPs cover the reported issue. Only cite sources you actually used.\n\n"
f"Operator Query: {state['query']}\n\n"
f"Equipment Analysis: {state['analysis']}\n\n"
f"Retrieved SOP Context:\n{context}"
)
report = structured_llm.invoke(prompt)
return {"report": report.model_dump()}
def validator_node(state: PipelineState) -> dict:
"""Validate the generated report for quality and confidence."""
report = state["report"]
retrieved_titles = [doc["metadata"]["title"] for doc in state["retrieved_docs"]]
issues = []
if report["confidence_score"] < 0.7:
issues.append(f"Low confidence score: {report['confidence_score']:.2f}")
for source in report["cited_sources"]:
if source not in retrieved_titles:
issues.append(f"Citation not found in retrieved docs: {source}")
if not report["recommended_actions"]:
issues.append("No recommended actions provided")
validation_passed = len(issues) == 0
retry_count = state["retry_count"] + (0 if validation_passed else 1)
return {
"validation_passed": validation_passed,
"retry_count": retry_count,
}Your four agents are ready. Next up, you will wire them into a LangGraph StateGraph with conditional routing so the validator can loop back to the generator when quality is low.
Assemble the LangGraph Pipeline and Run
Your four agent nodes are built and ready to work. Right now they exist as isolated functions with no way to communicate or pass state between each other.
In this step, you will wire the agents into a LangGraph StateGraph that orchestrates them in sequence. You will also add a conditional loop so the validator can reject low-confidence reports and send them back for regeneration.
In this step, get ready to:
- Wire all four agent nodes into a StateGraph with conditional routing.
- Create a CLI entry point that accepts operator queries and prints formatted incident reports.
- Test the full pipeline end-to-end with a sample query.
Build the conditional routing function
The validator agent sets a validation_passed flag on the pipeline state. A conditional edge reads that flag and decides the next destination: either finish the pipeline or loop back to the generator for another attempt.
- Create a new file called graph.py in your project folder.
- Add the imports and routing function by pasting this code:
"""LangGraph pipeline assembly and execution."""
from typing import Literal
from langgraph.graph import StateGraph, START, END
from state import PipelineState
from agents import query_analyzer_node, retriever_node, generator_node, validator_node
def route_after_validation(state: PipelineState) -> Literal["generator", "__end__"]:
"""Route based on validation result: retry generation or finish."""
if state["validation_passed"]:
return "__end__"
if state["retry_count"] >= 2:
return "__end__"
return "generator"What does this code do?
The route_after_validation function is the decision point at the end of the pipeline. LangGraph calls it after the validator runs to decide where to go next.
- If validation_passed is True, the report met quality thresholds, so we route to __end__ (finish).
- If retry_count >= 2, we have already retried twice, so we stop to avoid infinite loops.
- Otherwise, the report failed validation and we still have retries left, so we route back to "generator" for another attempt.
Wire the nodes into the StateGraph
With the routing logic defined, you can now register each agent as a node and connect them with edges. The edges define the execution order: query analysis first, then retrieval, then generation, then validation.
- Add the build_graph function below route_after_validation in graph.py:
def build_graph():
"""Build and compile the multi-agent LangGraph pipeline."""
workflow = StateGraph(PipelineState)
workflow.add_node("query_analyzer", query_analyzer_node)
workflow.add_node("retriever", retriever_node)
workflow.add_node("generator", generator_node)
workflow.add_node("validator", validator_node)
workflow.add_edge(START, "query_analyzer")
workflow.add_edge("query_analyzer", "retriever")
workflow.add_edge("retriever", "generator")
workflow.add_edge("generator", "validator")
workflow.add_conditional_edges("validator", route_after_validation)
return workflow.compile()What does this code do?
This function creates the entire agent pipeline as a compiled state machine.
- StateGraph(PipelineState) creates a new graph that uses your TypedDict as the shared state flowing between nodes.
- add_node registers each agent function as a named node in the graph.
- add_edge creates fixed connections: START flows to query_analyzer, then to retriever, then to generator, then to validator.
- add_conditional_edges connects the validator to your routing function, which picks the next node at runtime.
- workflow.compile() freezes the graph structure and returns a runnable object you can invoke with queries.
- Save graph.py.
Seeing an import error?
If your editor shows an error on the from langgraph.graph import StateGraph, START, END line, make sure you installed langgraph in your active Python environment. Run pip install -U langgraph to confirm it is installed.
Help me debug a LangGraph import error in my graph.py file.
✔️ Awesome, I've got everything!
Great. Double check you have saved graph.py.
ⓧ I'd like to double check the full code
"""LangGraph pipeline assembly and execution."""
from typing import Literal
from langgraph.graph import StateGraph, START, END
from state import PipelineState
from agents import query_analyzer_node, retriever_node, generator_node, validator_node
def route_after_validation(state: PipelineState) -> Literal["generator", "__end__"]:
"""Route based on validation result: retry generation or finish."""
if state["validation_passed"]:
return "__end__"
if state["retry_count"] >= 2:
return "__end__"
return "generator"
def build_graph():
"""Build and compile the multi-agent LangGraph pipeline."""
workflow = StateGraph(PipelineState)
workflow.add_node("query_analyzer", query_analyzer_node)
workflow.add_node("retriever", retriever_node)
workflow.add_node("generator", generator_node)
workflow.add_node("validator", validator_node)
workflow.add_edge(START, "query_analyzer")
workflow.add_edge("query_analyzer", "retriever")
workflow.add_edge("retriever", "generator")
workflow.add_edge("generator", "validator")
workflow.add_conditional_edges("validator", route_after_validation)
return workflow.compile()Create the CLI entry point
The pipeline is assembled but has no user interface. You need a way for operators to type queries and see formatted results. A simple CLI loop handles this perfectly for a local tool.
- Create a new file called main.py in your project folder.
- Add the imports and report formatting function by pasting this code:
"""CLI entry point for the Industrial Operations Copilot."""
import os
import json
from graph import build_graph
def format_report(result: dict) -> str:
"""Format the pipeline result as a readable incident report."""
report = result["report"]
analysis = result["analysis"]
output = []
output.append("=" * 60)
output.append("INCIDENT REPORT")
output.append("=" * 60)
output.append(f"Equipment: {analysis['equipment_id']}")
output.append(f"Urgency: {analysis['urgency'].upper()}")
output.append(f"Severity: {report['severity'].upper()}")
output.append(f"Confidence: {report['confidence_score']:.0%}")
output.append("-" * 60)
output.append(f"Root Cause: {report['root_cause']}")
output.append("-" * 60)
output.append("Recommended Actions:")
for i, action in enumerate(report["recommended_actions"], 1):
output.append(f" {i}. {action}")
output.append("-" * 60)
output.append("Cited Sources:")
for source in report["cited_sources"]:
output.append(f" - {source}")
output.append("-" * 60)
output.append(f"Validation: {'PASSED' if result['validation_passed'] else 'FAILED'}")
output.append(f"Retries: {result['retry_count']}")
output.append("=" * 60)
return "\n".join(output)What does this code do?
The format_report function takes the raw pipeline output and assembles a human-readable incident report.
- It pulls the equipment ID and urgency from the query analysis, and severity, root cause, and actions from the generated report.
- The confidence score displays as a percentage (e.g. 85%).
- Cited sources and validation status are printed at the bottom so the operator knows which SOPs were referenced.
- Add the main function and entry point below format_report in main.py:
def main():
"""Run the Industrial Operations Copilot CLI."""
# Check that the API key is available before starting
if not os.environ.get("GOOGLE_API_KEY"):
print("Error: Set GOOGLE_API_KEY environment variable.")
print("Get a free key at: https://aistudio.google.com/apikey")
return
print("Industrial Operations Copilot")
print("Type 'quit' to exit.\n")
# Compile the graph once, reuse for every query
graph = build_graph()
while True:
query = input("Operator Query> ").strip()
if query.lower() in ("quit", "exit", "q"):
break
if not query:
continue
print("\nProcessing query through agent pipeline...\n")
# Invoke the graph with initial state
result = graph.invoke({
"query": query,
"analysis": None,
"retrieved_docs": [],
"report": None,
"validation_passed": False,
"retry_count": 0,
})
print(format_report(result))
print()
if __name__ == "__main__":
main()What does this code do?
The main function is the interactive loop that drives the copilot.
- It first checks that GOOGLE_API_KEY is set, printing a helpful error if it is missing.
- build_graph() is called once to compile the pipeline. This avoids recompiling on every query.
- graph.invoke() runs the full agent pipeline with the initial state dict. LangGraph passes this state through each node sequentially.
- The if __name__ == "__main__" guard ensures the CLI only runs when you execute the file directly.
- Save main.py.
Seeing a ModuleNotFoundError?
If you get an error about graph or agents not being found, make sure all your files (models.py, state.py, ingest.py, agents.py, graph.py, main.py) are in the same directory. Run your command from that directory.
Help me fix a ModuleNotFoundError when running my main.py file.
✔️ Awesome, I've got everything!
Great. Double check you have saved main.py.
ⓧ I'd like to double check the full code
"""CLI entry point for the Industrial Operations Copilot."""
import os
import json
from graph import build_graph
def format_report(result: dict) -> str:
"""Format the pipeline result as a readable incident report."""
report = result["report"]
analysis = result["analysis"]
output = []
output.append("=" * 60)
output.append("INCIDENT REPORT")
output.append("=" * 60)
output.append(f"Equipment: {analysis['equipment_id']}")
output.append(f"Urgency: {analysis['urgency'].upper()}")
output.append(f"Severity: {report['severity'].upper()}")
output.append(f"Confidence: {report['confidence_score']:.0%}")
output.append("-" * 60)
output.append(f"Root Cause: {report['root_cause']}")
output.append("-" * 60)
output.append("Recommended Actions:")
for i, action in enumerate(report["recommended_actions"], 1):
output.append(f" {i}. {action}")
output.append("-" * 60)
output.append("Cited Sources:")
for source in report["cited_sources"]:
output.append(f" - {source}")
output.append("-" * 60)
output.append(f"Validation: {'PASSED' if result['validation_passed'] else 'FAILED'}")
output.append(f"Retries: {result['retry_count']}")
output.append("=" * 60)
return "\n".join(output)
def main():
"""Run the Industrial Operations Copilot CLI."""
if not os.environ.get("GOOGLE_API_KEY"):
print("Error: Set GOOGLE_API_KEY environment variable.")
print("Get a free key at: https://aistudio.google.com/apikey")
return
print("Industrial Operations Copilot")
print("Type 'quit' to exit.\n")
graph = build_graph()
while True:
query = input("Operator Query> ").strip()
if query.lower() in ("quit", "exit", "q"):
break
if not query:
continue
print("\nProcessing query through agent pipeline...\n")
result = graph.invoke({
"query": query,
"analysis": None,
"retrieved_docs": [],
"report": None,
"validation_passed": False,
"retry_count": 0,
})
print(format_report(result))
print()
if __name__ == "__main__":
main()Test the pipeline with a sample query
Everything is connected. Time to run the copilot and see all four agents work together on a real operator scenario.
- Run the copilot by executing this command in your terminal:
python main.pyYou should see the prompt Operator Query> waiting for your input.
- Type in this sample query and press Enter:
Pump P-401 showing high vibration and elevated temperatureThe pipeline processes for a few seconds as each agent runs in sequence. You should then see a structured incident report printed to the terminal.
What should I see?
The output is a formatted incident report with these sections:
- Equipment ID (P-401), urgency, and severity levels extracted by the query analyzer.
- A confidence score (displayed as a percentage). Anything above 70% means the validator approved the report.
- A root cause identified from the SOP documents (likely bearing failure based on combined vibration and temperature symptoms).
- Numbered recommended actions pulled directly from the relevant SOPs.
- Cited SOP titles showing which documents the generator referenced.
- Validation status (PASSED or FAILED) and retry count.
- Type quit and press Enter to exit the CLI.
Pipeline hanging or timing out?
The first run can take 10-20 seconds because Gemini processes structured output for multiple agents. If it hangs longer than 60 seconds, check that your GOOGLE_API_KEY environment variable is set correctly in the same terminal session.
If you see a validation FAILED result with retries, that means the validator caught an issue and the pipeline re-ran the generator. This is expected behavior for the conditional loop.
Help me debug why my multi-agent pipeline is timing out or returning errors.
Your multi-agent copilot is fully operational. Four specialized agents analyze, retrieve, generate, and validate structured incident reports from plain-language operator queries. Next up, you will extend this system with metrics tracking in the Secret Mission.
Secret mission
Add Agent Metrics and Evaluation Tracking
Your pipeline works, but you have no visibility into how it performs. Which agent is the bottleneck? How many tokens are you burning per query? Is confidence trending up or down over time? Add an observability layer that answers all three.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally with no cloud resources or paid services. There are no ongoing costs to worry about.
Decide whether to keep your project files for portfolio use or delete them to free up disk space.
Resources you used:
- Project folder on your Desktop containing all Python files (models.py, state.py, ingest.py, agents.py, graph.py, main.py).
- ./chroma_data directory containing ChromaDB vector store data (a few MB).
- metrics.json file storing pipeline performance metrics (if you completed the Secret Mission).
Keep everything running
No action needed. Choose this if you plan to keep building on your Industrial Operations Copilot or want it ready as a portfolio piece.
- Your project files stay on your Desktop, ready for future use.
- The ./chroma_data directory persists your SOP embeddings so you can query them again without re-ingesting.
- Your installed Python packages (langgraph, chromadb, langchain-google-genai) remain available for other projects.
Pause - I'll come back to this later
Since this project runs entirely locally with no background processes or cloud services, there is nothing to pause. Your files sit idle on disk until you return.
- Make sure you have exited the CLI by typing quit at the Operator Query> prompt.
- If you used a virtual environment, deactivate it by running deactivate in your terminal.
- Your project files and ChromaDB data remain on disk, ready when you come back.
Delete - I don't want to use this again
Remove all project files and data to start fresh. This deletes everything you built in this project.
- Delete the entire project folder from your Desktop by running this command (macOS/Linux):
cd ~/Desktop
rm -rf industrial-copilotOn Windows, run this instead:
cd ~\Desktop
rmdir /s /q industrial-copilotThis removes all Python files, the ./chroma_data directory, and metrics.json in one step.
- Optionally, uninstall the Python packages you installed for this project:
pip uninstall langgraph chromadb langchain-google-genai -y- Your Google AI API key remains active in Google AI Studio. You can revoke it at https://aistudio.google.com/apikey if you no longer need it.
Nice Work!
Nice Work!
You just built a production-grade multi-agent RAG system that takes plain-language operator queries and produces structured incident reports.
You've learned how to:
- Build a LangGraph StateGraph with four specialized agent nodes orchestrated in a sequential pipeline with conditional routing.
- Implement a RAG pipeline over industrial SOP documents using ChromaDB with metadata filtering and ranked retrieval.
- Produce structured Pydantic IncidentReport outputs with confidence scoring, source citations, and a validation loop that re-routes low-confidence outputs for regeneration.
- Secret Mission: Tracked per-agent latencies, token usage, and confidence score distributions for pipeline observability.
Ready to quiz yourself?