Mortgage Lending 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 MortgageSessionData:
    loan_number: Optional[str] = None
    customer_name: Optional[str] = None
    last_four_ssn: Optional[str] = None
    payment_amount: Optional[float] = None
    payment_method: Optional[str] = None
    questions_answered: list[str] = field(default_factory=list)
    moss_client: Optional[MossClient] = None

from livekit.agents import Agent, RunContext, function_tool
from moss import MossClient, QueryOptions

class MortgageRetrievalAgent(Agent):
    def __init__(self, moss_client: MossClient):
        self._moss = moss_client
        super().__init__(instructions="""
            You are a mortgage lending voice assistant.
            ALWAYS call search_mortgage_kb before answering any factual question.
            Keep answers short — this is voice, no bullet points or markdown.
            When the customer says they want to make a payment, call transfer_to_payment_flow.
        """)

    async def on_enter(self) -> None:
        await self.session.say(
            "Hi, this is Moss from mortgage services. I can help with questions "
            "about your loan, payment options, or rates. What can I help you with?"
        )

    @function_tool
    async def search_mortgage_kb(self, context: RunContext, question: str) -> str:
        """Search the mortgage knowledge base. Use for any factual question
        about loan products, eligibility, closing costs, or payment options."""
        results = await self._moss.query(
            "mortgage-lending-kb", question, QueryOptions(top_k=4, alpha=0.75)
        )
        if not results.docs:
            return "No relevant information found."
        data: MortgageSessionData = self.session.userdata
        data.questions_answered.append(question)
        return "\n".join(f"- {d.text}" for d in results.docs)

    @function_tool
    async def capture_loan_number(self, context: RunContext, loan_number: str) -> str:
        """Save the customer's loan number to session state."""
        data: MortgageSessionData = self.session.userdata
        data.loan_number = loan_number.strip()
        return f"Saved loan number {data.loan_number}."

    @function_tool
    async def transfer_to_payment_flow(self, context: RunContext) -> tuple:
        """Hand off to the payment flow agent when the customer wants to pay."""
        data: MortgageSessionData = self.session.userdata
        greeting = (
            "Got it. I have your loan number on file — connecting you to payments now."
            if data.loan_number
            else "Got it, let me hand you over to our payment flow."
        )
        return PaymentFlowAgent(), greeting

class PaymentFlowAgent(Agent):
    def __init__(self):
        super().__init__(instructions="""
            You are the payment flow agent. Steps in order:
            1. Read session state with read_session_state — skip any field already captured.
            2. Ask for last 4 SSN to verify identity. Call verify_identity.
            3. Ask for the payment amount. Call set_payment_amount.
            4. Ask for the payment method (bank transfer, debit card, autopay).
               Call set_payment_method.
            5. Read back all four facts and ask for confirmation.
            6. On confirmation, call submit_payment.
            Never ask for full SSN or full card numbers. Last 4 only.
            If the customer asks a mortgage question, call return_to_advisor.
        """)

    async def on_enter(self) -> None:
        data: MortgageSessionData = self.session.userdata
        if data.loan_number:
            await self.session.say(
                f"Hi, I'll get your payment set up. I have loan number "
                f"{data.loan_number} on file — is that the one you want to pay?"
            )
        else:
            await self.session.say("Hi, I'll get your payment set up. What's your loan number?")

    @function_tool
    async def read_session_state(self, context: RunContext) -> str:
        """Return what's already known about the customer this call."""
        data: MortgageSessionData = self.session.userdata
        known = {k: v for k, v in {
            "loan_number": data.loan_number,
            "last_four_ssn": data.last_four_ssn,
            "payment_amount": data.payment_amount,
            "payment_method": data.payment_method,
        }.items() if v}
        return ", ".join(f"{k}={v}" for k, v in known.items()) or "nothing on file yet"

    @function_tool
    async def verify_identity(self, context: RunContext, last_four_ssn: str) -> str:
        """Save the last four digits of SSN for verification."""
        digits = "".join(c for c in last_four_ssn if c.isdigit())
        if len(digits) != 4:
            return "Please ask the customer to repeat the last four digits clearly."
        self.session.userdata.last_four_ssn = digits
        return "Identity captured."

    @function_tool
    async def set_payment_amount(self, context: RunContext, amount: float) -> str:
        """Record the payment amount in dollars."""
        self.session.userdata.payment_amount = amount
        return f"Recorded ${amount:,.2f}."

    @function_tool
    async def set_payment_method(self, context: RunContext, method: str) -> str:
        """Record the payment method."""
        self.session.userdata.payment_method = method.strip().lower()
        return f"Recorded {method}."

    @function_tool
    async def submit_payment(self, context: RunContext) -> str:
        """Submit the payment after customer confirmation."""
        data: MortgageSessionData = self.session.userdata
        confirmation = f"MOSS-{abs(hash(data.loan_number)) % 10_000_000:07d}"
        return (
            f"Payment of ${data.payment_amount:,.2f} submitted via "
            f"{data.payment_method}. Confirmation number {confirmation}."
        )

    @function_tool
    async def return_to_advisor(self, context: RunContext) -> tuple:
        """Hand back to the retrieval agent for mortgage questions."""
        data: MortgageSessionData = self.session.userdata
        return MortgageRetrievalAgent(data.moss_client), "Sure, let me get you back to the advisor."

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"])
    await moss_client.load_index("mortgage-lending-kb")

    session = AgentSession[MortgageSessionData](
        userdata=MortgageSessionData(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=MortgageRetrievalAgent(moss_client), room=ctx.room)

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

python agent.py console