01 · Ambiguous Business Names
Problem: Users sometimes provided unclear campaign names instead of stable IDs.
Solution: I added context validation and governed entity resolution. The Agent asks for clarification only when needed.
The system has two controlled answer paths. RAG explains knowledge. Analytics returns trusted business numbers. LangGraph connects the modules.
Three practical questions interviewers may ask about developing this AI Agent.
Problem: Users sometimes provided unclear campaign names instead of stable IDs.
Solution: I added context validation and governed entity resolution. The Agent asks for clarification only when needed.
Problem: Free-form LLM outputs could produce inconsistent metrics or unsafe queries.
Solution: I separated intent understanding from business validation and used predefined SQL controlled by application code.
Problem: Repeated manual data requests made reservation analysis slower and harder to standardize.
Solution: I built a governed Reservation Data Mart and connected the Agent to its predefined metrics through QueryBackend.
from typing import Literal
from pydantic import BaseModel, Field
class ReservationQuery(BaseModel):
country: str | None = None
country_code: str | None = None
product: str | None = None
product_id: str | None = None
campaign_name: str | None = None
campaign_id: str | None = None
campaign_month: int | None = None
campaign_year: int | None = None
class ExtractedRequest(BaseModel):
intent: Literal["knowledge", "analytics"]
metric: str = "summary"
detail_requested: bool = False
query: ReservationQuery = Field(default_factory=ReservationQuery)
request = structured_llm.invoke(prompt)
state["query"] = request.query.model_dump()class AgentState(TypedDict, total=False):
question: str
intent: str
metric: str
detail_requested: bool
query: dict
resolved_context: dict
status: str
answer: str
route: str
def extract_node(state: AgentState) -> AgentState:
request = extractor.extract(state["question"])
return {
**state,
"intent": request.intent,
"metric": request.metric,
"detail_requested": request.detail_requested,
"query": request.query.model_dump(),
}
def after_extract(state: AgentState) -> str:
if state["intent"] == "knowledge":
return "knowledge"
return "validate"
graph.add_edge(START, "extract")
graph.add_conditional_edges(
"extract",
after_extract,
{
"knowledge": "knowledge",
"validate": "validate",
},
)Business definitions, metric rules and Data Mart knowledge.
Workflow routing and shared state.
Document loading, indexing and RAG query engine.
Stores embedding vectors and performs similarity search.
Exposes the Agent as an HTTP service.
The AI Agent is an interface on top of governed business data. It does not replace the Data Mart.
Grain: User × Campaign × Product × Site
Stable context: campaign, product, country / site IDs
Business logic: governed metrics and the reserved-but-not-ordered tag
Campaign dimension
Product dimension
Country / Site dimension
Use this mental model: File → Document → Chunks → Embeddings → FAISS → Top-K → LLM.
LlamaIndex's standard representation of a loaded file. It contains the text plus metadata such as the file name.
The embedding model returns 1536 numbers for each text embedding. The FAISS index must use the same dimension.
Flat = exact search over stored vectors. L2 = Euclidean distance. Smaller distance means more similar.
Tells LlamaIndex where the vector embeddings should be stored.
Retrieve the three most relevant chunks for each user question.
app/rag/ingestion/loader.pyapp/rag/embeddings/embedder.pyapp/rag/vectorstore/faiss_store.pyapp/rag/retrieval/retriever.pyapp/rag/service.pyanswer(question) interfaceclass AgentState(TypedDict, total=False):
"""Shared working state passed through the LangGraph workflow."""
question: str
intent: str
metric: str
detail_requested: bool
query: dict
# Resolved Campaign + Product + Country context.
resolved_context: dict
# validated / resolved / clarification / not_found / answered
status: str
# Final user-facing response.
answer: str
# knowledge / analytics
route: str
def extract_node(self, state: AgentState) -> AgentState:
request = self.extractor.extract(state["question"])
return {
**state,
"intent": request.intent,
"metric": request.metric,
"detail_requested": request.detail_requested,
"query": request.query.model_dump(),
}
def validate_node(self, state: AgentState) -> AgentState:
query = ReservationQuery(**state["query"])
missing = missing_analytics_context(query)
if missing:
return {
**state,
"status": "clarification",
"route": "analytics",
"answer": "Please provide " + ", ".join(missing) + ". I will not guess.",
}
return {**state, "status": "validated"}
def resolve_node(self, state: AgentState) -> AgentState:
campaigns = self.resolver.resolve(
ReservationQuery(**state["query"])
)
if not campaigns:
return {
**state,
"status": "not_found",
"answer": "No campaign matched the supplied business context.",
}
if len(campaigns) > 1:
choices = "; ".join(
f"{item.campaign_id} — {item.campaign_name}"
for item in campaigns
)
return {
**state,
"status": "clarification",
"answer": "Multiple campaigns matched. Choose one campaign_id: " + choices,
}
return {
**state,
"resolved_context": campaigns[0].model_dump(),
"status": "resolved",
}
| Node | Responsibility |
|---|---|
extract_node | Natural language → structured request |
knowledge_node | Call KnowledgeRAG for definitions / metadata |
validate_node | Check required business context |
resolve_node | Resolve exactly one campaign context |
analytics_node | Run controlled analytics SQL |
graph.add_edge(START, "extract")graph.add_conditional_edges(
"extract",
self._after_extract,
{
"knowledge": "knowledge",
"validate": "validate",
},
)@staticmethod
def _after_extract(state: AgentState) -> str:
return "knowledge" if state["intent"] == "knowledge" else "validate"
@staticmethod
def _after_validate(state: AgentState) -> str:
return "end" if state["status"] == "clarification" else "resolve"
@staticmethod
def _after_resolve(state: AgentState) -> str:
return "analytics" if state["status"] == "resolved" else "end"
def _build_langgraph(self):
from langgraph.graph import END, START, StateGraph
graph = StateGraph(AgentState)
graph.add_node("extract", self.extract_node)
graph.add_node("knowledge", self.knowledge_node)
graph.add_node("validate", self.validate_node)
graph.add_node("resolve", self.resolve_node)
graph.add_node("analytics", self.analytics_node)
graph.add_edge(START, "extract")
graph.add_conditional_edges(
"extract",
self._after_extract,
{
"knowledge": "knowledge",
"validate": "validate",
},
)
graph.add_edge("knowledge", END)
graph.add_conditional_edges(
"validate",
self._after_validate,
{
"resolve": "resolve",
"end": END,
},
)
graph.add_conditional_edges(
"resolve",
self._after_resolve,
{
"analytics": "analytics",
"end": END,
},
)
graph.add_edge("analytics", END)
return graph.compile()
def invoke(self, question: str) -> AgentState:
state: AgentState = {"question": question}
if self._graph is not None:
return self._graph.invoke(state)
# Fallback path when LangGraph is not installed.
state = self.extract_node(state)
if self._after_extract(state) == "knowledge":
return self.knowledge_node(state)
state = self.validate_node(state)
if self._after_validate(state) == "end":
return state
state = self.resolve_node(state)
if self._after_resolve(state) == "end":
return state
return self.analytics_node(state)
{question}intent / metric / queryresolved_contextstatus=answered
answer=...{
"question": "How many users reserved Phone Mi 17 Pro in Germany for CMP001?",
"intent": "analytics",
"metric": "reserved_users",
"detail_requested": false,
"query": {
"country": "Germany",
"product": "Phone Mi 17 Pro",
"campaign_id": "CMP001"
},
"resolved_context": {
"campaign_id": "CMP001",
"product_id": "P001",
"country_code": "DE"
},
"status": "answered",
"route": "analytics",
"answer": "..."
}
The analytics path converts a business plan into a stable governed context, then executes allowlisted SQL.
clarification and not_found stop before SQL. Only resolved enters the analytics node. conflict and unsupported are not implemented in this version.Validate directly against governed dimension tables. Do not ask the LLM to resolve the ID again.
Search governed candidates and resolve the name only when needed.
Return candidate IDs and ask the user to retry. No session-resume loop in this simple version.
These two examples match the presentation: one proves the Knowledge RAG path; the other proves the controlled Analytics path.
The router sends the question to Knowledge RAG.
Answer: reserved_users means distinct users with freserve_flag = 1, based on trusted metric knowledge.
CMP001 → exact validation
Germany → DE through governed name resolution
reserved_users → allowlisted controlled SQL
Return the governed metric result.
Top level is organized by system capability; each capability is internally organized by responsibility.
models = typed data; validation = enough context; resolution = governed IDs; metrics = allowlist; query = backend adapter; service = orchestration.
nodes = one responsibility per step; state = shared working data; workflow = conditional routing; api = service entry; llm = shared model client.
1. graph/workflow.py
2. graph/nodes/extract.py
3. rag/service.py
4. rag/*
5. analytics/resolution/
6. analytics/metrics/
7. analytics/service.py
What is running in production, and how does one user request reach the trusted answer?
Reference: Production Deploy, Slide 8. Ingress and Service route traffic to the application Pod. The application calls an external LLM API and the enterprise Query Engine; it does not host model weights.
RAG retrieves grounded definitions using LlamaIndex and FAISS. See Tab 3 for details.
Validate the required context → Resolve entities → execute controlled SQL when clear. Ask for clarification only if information is missing or ambiguous. See Tab 5.
API handling and structured request validation.
Workflow state and routing to Knowledge RAG or Analytics.
External model services; calling an LLM API is not the same as deploying an LLM model.
Runs allowlisted business queries against the governed Reservation Data Mart.
ConfigMap: non-sensitive runtime settings. Secrets: sensitive LLM and data-platform credentials. Monitoring: latency, errors, health, and traces. Deployment commands belong to Tab 9.
How to package, test, deploy, observe, and roll back the Python AI Agent. These commands are learning examples, not unverified claims about the production CI/CD provider.
Remember: Production (Tab 8) explains how requests run. Deployment (this tab) explains how engineers ship the application.
A practical repository separates API entry points, Agent logic, knowledge retrieval, analytics, tests, and deployment manifests.
reservation-agent/
├── app/
│ ├── api/main.py
│ ├── graph/workflow.py
│ ├── rag/retriever.py
│ ├── analytics/query_backend.py
│ └── llm/client.py
├── tests/
├── requirements.txt
├── Dockerfile
├── .dockerignore
├── k8s/
│ ├── deployment.yaml
│ ├── service.yaml
│ ├── ingress.yaml
│ └── configmap.yaml
└── .github/workflows/ci-cd.yml
# Example only: adapt to your actual FastAPI module.
from fastapi import FastAPI
app = FastAPI()
@app.get("/healthz")
def healthz():
return {"status": "ok"}
@app.get("/readyz")
def readyz():
# Add lightweight readiness checks if required.
return {"ready": True}
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
uvicorn app.api.main:app --host 127.0.0.1 --port 8000
curl http://127.0.0.1:8000/healthz
Docker builds an immutable application image. Do not bake secrets into the image. Versioned tags make deployment and rollback traceable.
FROM python:3.11-slim
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app/ ./app/
EXPOSE 8000
CMD ["uvicorn", "app.api.main:app", "--host", "0.0.0.0", "--port", "8000"]
# .dockerignore
.git
.venv
__pycache__
.pytest_cache
.env
*.pyc
docker build -t reservation-agent:v1 .
docker run --rm -p 8000:8000 reservation-agent:v1
# In another terminal:
curl http://localhost:8000/healthz
# Examine runtime output:
docker ps
docker logs <container-id>
Continuous Integration (CI) starts after a code push or pull request. Tests should exercise both ordinary Python code and Agent-specific guardrails.
Valid and invalid Pydantic extraction outputs.
Relevant retrieval, empty search, grounded answers.
Campaign validation, entity resolution, approved metric keys, no unsafe free-form SQL.
python -m pip install -r requirements.txt
python -m pip install pytest ruff
ruff check app tests
pytest -q
docker build -t reservation-agent:test .
The registry hosts built images so the Kubernetes cluster can pull them. Choose the registry actually provided by your organization.
# Example registry; replace with your authorized registry.
export REGISTRY=registry.example.com/team
export IMAGE=$REGISTRY/reservation-agent
export TAG=v1
docker tag reservation-agent:v1 "$IMAGE:$TAG"
docker login registry.example.com
docker push "$IMAGE:$TAG"
Deployment manages replica Pods. Service provides internal access. Ingress routes external HTTPS requests to that Service when an Ingress controller and TLS are configured.
# k8s/deployment.yaml — illustrative manifest
apiVersion: apps/v1
kind: Deployment
metadata:
name: reservation-agent
spec:
replicas: 2
strategy:
type: RollingUpdate
selector:
matchLabels:
app: reservation-agent
template:
metadata:
labels:
app: reservation-agent
spec:
containers:
- name: api
image: registry.example.com/team/reservation-agent:v1
ports:
- containerPort: 8000
envFrom:
- configMapRef:
name: reservation-agent-config
- secretRef:
name: reservation-agent-secrets
readinessProbe:
httpGet:
path: /readyz
port: 8000
initialDelaySeconds: 5
periodSeconds: 10
livenessProbe:
httpGet:
path: /healthz
port: 8000
initialDelaySeconds: 15
periodSeconds: 20
resources:
requests:
cpu: 100m
memory: 256Mi
limits:
cpu: 1000m
memory: 1Gi
# k8s/service.yaml
apiVersion: v1
kind: Service
metadata:
name: reservation-agent
spec:
selector:
app: reservation-agent
ports:
- port: 80
targetPort: 8000
type: ClusterIP
# k8s/ingress.yaml — controller/TLS must exist
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: reservation-agent
spec:
ingressClassName: nginx
tls:
- hosts: [agent.example.com]
secretName: agent-tls
rules:
- host: agent.example.com
http:
paths:
- path: /
pathType: Prefix
backend:
service:
name: reservation-agent
port:
number: 80
Keep non-sensitive runtime settings separate from credentials. The app sends controlled SQL through QueryBackend → Data Platform / Query Engine → Reservation Data Mart.
# k8s/configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
name: reservation-agent-config
data:
REGION: "sg"
QUERY_BACKEND: "enterprise-query-gateway"
LOG_LEVEL: "INFO"
# Illustration only: create a secret from secured inputs.
# In production, use your organization's approved secret manager.
kubectl create secret generic reservation-agent-secrets \
--from-literal=LLM_API_KEY='REPLACE_IN_SECURE_ENV' \
--from-literal=DATA_PLATFORM_TOKEN='REPLACE_IN_SECURE_ENV' \
--dry-run=client -o yaml | kubectl apply -f -
Continuous Delivery / Deployment (CD) promotes an approved image to Kubernetes, checks its rollout, and supports rollback. The exact approval policy depends on the team.
# Prerequisites: correct kube context, permissions, and manifests.
kubectl config current-context
kubectl apply -f k8s/configmap.yaml
kubectl apply -f k8s/deployment.yaml
kubectl apply -f k8s/service.yaml
kubectl apply -f k8s/ingress.yaml
kubectl set image deployment/reservation-agent \
api=registry.example.com/team/reservation-agent:v2
kubectl rollout status deployment/reservation-agent
kubectl get pods -l app=reservation-agent
kubectl logs deployment/reservation-agent --tail=100
# If the release fails:
kubectl rollout undo deployment/reservation-agent
# Illustrative GitHub Actions flow, NOT verified project CI/CD.
name: Test and build
on: [push, pull_request]
jobs:
test-and-build:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- run: pip install -r requirements.txt pytest ruff
- run: ruff check app tests
- run: pytest -q
- run: docker build -t reservation-agent:ci .
# Production registry publishing and Kubernetes CD
# require separately configured identities and approvals.
After release, the business runtime is different from the CI/CD path. Users enter through Feishu; the FastAPI service forwards requests to LangGraph; RAG and Analytics use external services.
kubectl get deployments,services,ingresses
kubectl describe deployment reservation-agent
kubectl top pods
kubectl logs deployment/reservation-agent --since=15m
kubectl scale deployment/reservation-agent --replicas=3
kubectl get pods -l app=reservation-agent -w
Request count, error rate, p95 latency, and correlation IDs.
LLM timeouts, retrieval latency, entity-resolution failures, and query duration.
Pod restarts, readiness, CPU / memory, and rollout status.
kubectl get pods
kubectl describe pod <pod-name>
kubectl logs <pod-name> --previous
kubectl get events --sort-by=.lastTimestamp
kubectl rollout history deployment/reservation-agent
kubectl rollout undo deployment/reservation-agent
Diagnosis sequence: check user-facing symptom → trace FastAPI request → inspect Agent state or LLM API → inspect RAG / QueryBackend errors → check Kubernetes Pod health → roll back a bad release if needed.
Every English sentence below is taken from the exact Speaker Notes in PPT v106. Only line breaks and sentence display are adjusted for click-to-listen practice.