Building an AI-Powered Customer Response Automation System with Datazone

[
  {
    "mail": "john.smith@gmail.com",
    "subject": "Card Not Working",
    "message_content": "Hello, my bank card has balance but it didn't work when I tried to make a purchase.",
    "source": "email",
    "timestamp": "2025-08-01T09:15:30Z"
  }
]

class SentimentAnalysisResultFormat(BaseModel):
    issue_title: str = Field(description="Brief title for the issue")
    summary: str = Field(description="Summary of the message")
    response: str = Field(description="Response to the message")
    solving_score: int = Field(description="Confidence score, 1-10")
    request_type: Literal["information_request", "complaint", "suggestion"] = Field()
    sentiment_category: Literal["positive", "negative", "neutral"] = Field()
    urgency_level: int = Field(description="Urgency level, 1-10")

agent = Agent(
    model=AnthropicModel(
        provider=AnthropicProvider(api_key=ANTHROPIC_API_KEY),
        model_name="claude-3-7-sonnet-latest",
    ),
    output_type=SentimentAnalysisResultFormat,
    system_prompt=SYSTEM_PROMPT,
)

You are an automation agent for responding to user feedback.
...
Scale the solving_score as follows:
    - 1-3: Not confident. Use when you're unsure.
    - 4-6: Moderately confident. Use for complaints needing more info.
    - 7-8: Confident. Use for basic information requests.
    - 9-10: Very confident. Use for simple suggestions or appreciation.

@transform(description="Reads emails from file", engine="pandas")
def read_emails():
    with open("sample_data.json", "r") as file:
        data = json.load(file)
    return pd.DataFrame(data)

@transform(
    input_mapping={"emails": Input(read_emails)},
    engine="pandas"
)
def analyze_emails(emails):
    for index, row in emails.iterrows():
        user_message = f"Email: {row['mail']}\nSubject: {row['subject']}\nContent: {row['message_content']}"
        
        # Process with AI
        result = agent.run_sync(user_message)
        
        # Store analysis results
        emails.at[index, "issue_title"] = result.output.issue_title
        emails.at[index, "sentiment_category"] = result.output.sentiment_category
        emails.at[index, "solving_score"] = result.output.solving_score
        emails.at[index, "response"] = result.output.response
        emails.at[index, "status"] = "ANALYZED"
    
    return emails

@transform(
    input_mapping={"analyzed_emails": Input(analyze_emails)},
    engine="pandas"
)
def handle_emails(analyzed_emails):
    # Determine which emails need human review
    for index, row in analyzed_emails.iterrows():
        if row["solving_score"] < 8:
            analyzed_emails.at[index, "status"] = "NEED_ACTION"
    
    return analyzed_emails

# If the AI confidence is below our threshold, route to human team
if row["solving_score"] < 8:
    # Mark for human review
    emails.at[index, "status"] = "NEED_ACTION"

{
"mail": "john.smith@gmail.com",
"subject": "Card Not Working",
"message_content": "Hello, my bank card has balance but it didn't work when I tried to make a purchase.",
"source": "email",
"timestamp": "2025-08-01T09:15:30Z",
"issue_title": "Bank Card Payment Failure",
"sentiment_category": "negative",
"solving_score": 7,
"response": "I'm sorry to hear about the trouble with your bank card. This could be due to several reasons such as a temporary system issue, a hold on your account, or a security feature triggered by unusual purchase activity. Please try the following steps:\n\n1. Try the card again at a different terminal\n2. Contact your bank's customer service at the number on the back of your card\n3. Check your online banking to ensure there are no restrictions on your account\n\nIf you continue to experience issues, please reply with more details about where you attempted to make the purchase and we'll investigate further.",
"status": "NEED_ACTION"
}