import os
from livekit.agents import Agent, AgentSession, RunContext, function_tool
from moss import MossClient, QueryOptions
JOB_INDEX = os.getenv("MOSS_JOB_INDEX_NAME", "job-senior-backend-payments")
CANDIDATE_INDEX = os.getenv("MOSS_CANDIDATE_INDEX_NAME", "candidate-strong-match")
class ScreeningAgent(Agent):
def __init__(self, moss_client: MossClient):
self._moss = moss_client
super().__init__(instructions="""
You are a voice screening interviewer. You have two retrieval tools:
- lookup_job_requirement — searches the JOB DESCRIPTION
- lookup_resume_fact — searches the CANDIDATE RESUME
Ground every factual statement in tool output. Never invent requirements,
compensation, team details, or claims about the candidate.
Run a 5-phase interview: intro/consent → background → role-fit →
candidate Q&A → close. Capture rubric scores with record_rubric_entry.
Bias rules (these override everything else): do NOT ask about or infer
age, marital status, family plans, religion, national origin, or disability.
If the candidate volunteers any of these, acknowledge briefly and move on.
Voice style: one question at a time, allow silence, keep replies short.
""")
async def on_enter(self) -> None:
# Pre-fetch role context before the first word
role_context = await self._query(JOB_INDEX, "role title, company name, team", "JD")
await self.session.generate_reply(
instructions=(
"Greet the candidate warmly. Name the role, company, and team "
"using ONLY the context below — do not invent any detail. "
"Explain this is a ~25-minute recorded screening and ask for consent.\n\n"
f"Role context:\n{role_context}"
),
)
@function_tool
async def lookup_job_requirement(self, context: RunContext, query: str) -> str:
"""Search the job description for requirements, team info, comp, and process.
Use before making any statement about the role or answering a candidate question."""
return await self._query(JOB_INDEX, query, "JD")
@function_tool
async def lookup_resume_fact(self, context: RunContext, query: str) -> str:
"""Search the candidate's resume for projects, skills, and experience.
Use before asking a follow-up so the question is specific, not generic."""
return await self._query(CANDIDATE_INDEX, query, "Resume")
async def _query(self, index: str, query: str, source: str) -> str:
results = await self._moss.query(index, query, QueryOptions(top_k=4, alpha=0.75))
if not results.docs:
return f"No relevant {source.lower()} content found."
return "\n".join(f"- {d.text}" for d in results.docs)
@function_tool
async def record_consent(self, context: RunContext, consented: bool) -> str:
"""Record consent to be recorded. Call immediately after asking. End if declined."""
self.session.userdata.consent_to_record = consented
return "Consent captured." if consented else "Consent declined; end the screening."
@function_tool
async def record_rubric_entry(
self, context: RunContext, skill: str, score: int, evidence: str
) -> str:
"""Record one rubric row. score: 1=no signal, 3=competent, 5=strong.
evidence: brief paraphrase of what the candidate said."""
if not 1 <= score <= 5:
return "Score must be 1–5."
self.session.userdata.rubric[skill] = RubricEntry(
score=score, evidence=evidence.strip(), skill=skill
)
return f"Recorded {skill}={score}."
@function_tool
async def record_candidate_question(
self, context: RunContext, question: str, answer_summary: str
) -> str:
"""Log a question the candidate asked during Q&A."""
self.session.userdata.candidate_questions.append(
CandidateQuestion(question=question.strip(), answer_summary=answer_summary.strip())
)
return "Question logged."
@function_tool
async def submit_scorecard(self, context: RunContext) -> str:
"""Write the final scorecard JSON. Call once at the end of the screening."""
data: ScreeningSessionData = self.session.userdata
if data.consent_to_record is not True:
return "Cannot submit a scorecard without recorded consent."
scorecard = _build_scorecard(data)
# Write to disk (replace with your own storage in production)
import json
from pathlib import Path
path = Path("./scorecards") / f"{data.candidate_id}.json"
path.parent.mkdir(exist_ok=True)
path.write_text(json.dumps(scorecard, indent=2) + "\n", encoding="utf-8")
return f"Scorecard written. Tell the candidate the team reviews within 3 business days."
@function_tool
async def end_screening(self, context: RunContext, reason: str) -> str:
"""End the screening immediately. Use only if consent was declined."""
return "Thank the candidate politely and stop."
