Candidate Screening Agent

pip install "livekit-agents>=1.0.0" \
  livekit-plugins-openai livekit-plugins-deepgram \
  livekit-plugins-silero livekit-plugins-cartesia \
  moss python-dotenv

from dataclasses import dataclass, field
from typing import Optional
from moss import MossClient

@dataclass
class RubricEntry:
    score: int       # 1–5: 1=no signal, 3=competent, 5=strong
    evidence: str    # candidate's words, briefly paraphrased
    skill: str       # JD skill tag e.g. "postgres", "payments_domain"

@dataclass
class CandidateQuestion:
    question: str
    answer_summary: str

@dataclass
class ScreeningSessionData:
    candidate_id: str
    role_id: str
    consent_to_record: Optional[bool] = None
    rubric: dict[str, RubricEntry] = field(default_factory=dict)
    candidate_questions: list[CandidateQuestion] = field(default_factory=list)
    notes: list[str] = field(default_factory=list)
    moss_client: Optional[MossClient] = None

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."

def _recommendation_from_rubric(rubric: dict) -> str:
    if not rubric:
        return "no_signal"
    scores = [e.score for e in rubric.values()]
    avg = sum(scores) / len(scores)
    low_count = sum(1 for s in scores if s <= 2)
    if avg >= 4.0 and low_count == 0:
        return "advance_to_technical"
    if avg >= 3.0 and low_count <= 1:
        return "borderline_review"
    return "do_not_advance"

def _build_scorecard(data: ScreeningSessionData) -> dict:
    return {
        "candidate_id": data.candidate_id,
        "role_id": data.role_id,
        "rubric": {
            skill: {"score": e.score, "evidence": e.evidence}
            for skill, e in data.rubric.items()
        },
        "candidate_questions": [
            {"question": q.question, "answer_summary": q.answer_summary}
            for q in data.candidate_questions
        ],
        "notes": data.notes,
        "recommendation": _recommendation_from_rubric(data.rubric),
        "schema_version": 1,
    }

import os
from livekit.agents import AgentSession, JobContext, WorkerOptions, cli
from livekit.plugins import cartesia, deepgram, openai, silero
from moss import MossClient

async def entrypoint(ctx: JobContext):
    await ctx.connect()

    moss_client = MossClient(os.environ["MOSS_PROJECT_ID"], os.environ["MOSS_PROJECT_KEY"])
    for index in (JOB_INDEX, CANDIDATE_INDEX):
        await moss_client.load_index(index)

    session = AgentSession[ScreeningSessionData](
        userdata=ScreeningSessionData(
            candidate_id=os.getenv("SCREENING_CANDIDATE_ID", "candidate"),
            role_id=os.getenv("SCREENING_ROLE_ID", "role"),
            moss_client=moss_client,
        ),
        stt=deepgram.STT(model="nova-2"),
        llm=openai.LLM(model="gpt-4o"),
        tts=cartesia.TTS(),
        vad=silero.VAD.load(),
    )
    await session.start(agent=ScreeningAgent(moss_client), room=ctx.room)

if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))

python agent.py console