Spaces:
Sleeping
Sleeping
File size: 2,605 Bytes
f5ba3d2 4f164e3 f5ba3d2 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 |
from typing import TypedDict, List, Optional, Annotated, Union
from operator import add
from pydantic import BaseModel, Field
# --- 1. API Input/Output Models (Strictly Problem Statement 2) ---
class Message(BaseModel):
"""Represents a single message in the conversation."""
sender: str = Field(..., description="scammer or user")
text: str = Field(..., description="Message content")
# Flexible timestamp to handle both ISO-8601 strings (PDF) and Unix integers (Tester)
timestamp: Union[str, int] = Field(..., description="ISO-8601 format or Unix timestamp")
class Metadata(BaseModel):
"""Optional metadata about the conversation channel."""
channel: Optional[str] = Field(None, description="SMS / WhatsApp / Email / Chat")
language: Optional[str] = Field(None, description="Language used")
locale: Optional[str] = Field(None, description="Country or region")
class HoneypotRequest(BaseModel):
"""The incoming request body for the honeypot API."""
sessionId: str = Field(..., description="Unique session ID")
message: Message = Field(..., description="The latest incoming message")
conversationHistory: List[Message] = Field(default_factory=list, description="All previous messages")
metadata: Optional[Metadata] = None
class ExtractedIntelligence(BaseModel):
"""Structured data to be extracted from the conversation."""
bankAccounts: List[str] = Field(default_factory=list)
upiIds: List[str] = Field(default_factory=list) # Strictly 'upiIds' as per PDF
phishingLinks: List[str] = Field(default_factory=list)
phoneNumbers: List[str] = Field(default_factory=list)
suspiciousKeywords: List[str] = Field(default_factory=list)
unknownNumbers: List[str] = Field(default_factory=list)
class EngagementMetrics(BaseModel):
"""Metrics for the engagement."""
engagementDurationSeconds: int
totalMessagesExchanged: int
class HoneypotResponse(BaseModel):
"""The outgoing response body from the honeypot API."""
status: str = Field(..., description="success")
scamDetected: bool
engagementMetrics: EngagementMetrics
extractedIntelligence: ExtractedIntelligence
agentNotes: str
# --- 2. LangGraph State Model ---
class AgentState(TypedDict):
"""The state object for the LangGraph state machine."""
sessionId: str
conversationHistory: Annotated[List[Message], add]
scamDetected: bool
extractedIntelligence: ExtractedIntelligence
agentNotes: str
totalMessagesExchanged: int
should_continue_engagement: bool
agent_response_text: str
callbackSent: bool
|