124 lines
5 KiB
Python
Executable file
124 lines
5 KiB
Python
Executable file
from app.models.schemas import SubReview, RagStatus, OverallStatus
|
|
from app.services.gemini_service import GeminiService
|
|
|
|
|
|
class LeadAgent:
|
|
"""
|
|
Lead Agent - synthesizes specialist agent reviews into final verdict.
|
|
|
|
Applies the decision logic:
|
|
- Financial promotion detected → Requires Manual Legal Review
|
|
- Any Error status → Analysis Error
|
|
- Any Red status → Failed
|
|
- Otherwise → Passed
|
|
"""
|
|
|
|
name = "Lead Agent"
|
|
|
|
def __init__(self, gemini_service: GeminiService):
|
|
"""
|
|
Initialize the Lead Agent.
|
|
|
|
Args:
|
|
gemini_service: Service for making Gemini API calls (for summary generation)
|
|
"""
|
|
self.gemini = gemini_service
|
|
|
|
async def synthesize(
|
|
self,
|
|
reviews: dict[str, SubReview],
|
|
) -> tuple[OverallStatus, str, str | None]:
|
|
"""
|
|
Synthesize specialist reviews into final verdict and summary.
|
|
|
|
Args:
|
|
reviews: Dictionary mapping agent names to their SubReview results
|
|
|
|
Returns:
|
|
Tuple of (overall_status, summary, financial_promotion_reason)
|
|
"""
|
|
legal_review = reviews.get("Legal Agent")
|
|
|
|
# Check for financial promotion (from Legal Agent)
|
|
is_financial_promotion = (
|
|
legal_review is not None
|
|
and legal_review.isFinancialPromotion is True
|
|
)
|
|
financial_promotion_reason = (
|
|
legal_review.financialPromotionReason
|
|
if legal_review and legal_review.financialPromotionReason
|
|
else None
|
|
)
|
|
|
|
# Build the prompt for Gemini to generate summary
|
|
prompt = f"""
|
|
You are a Lead Agent responsible for auditing a marketing proof. You have received feedback from specialist AI agents.
|
|
Your task is to provide a final verdict and write a concise, professional summary to the user.
|
|
|
|
Here is the logic you must follow:
|
|
1. The Legal Agent has determined if this is a financial promotion: {is_financial_promotion}.
|
|
2. If it IS a financial promotion, the final verdict MUST be 'Requires Manual Legal Review'. Your summary should state this clearly, explain that a separate manual legal review is required, and then summarize any other issues found by the other agents.
|
|
3. If it is NOT a financial promotion, follow the standard logic:
|
|
a. If ANY specialist agent reports a 'ragStatus' of 'Error', the final verdict MUST be 'Analysis Error'.
|
|
b. If ANY specialist agent reports a 'Red' status (and there are no 'Error' statuses), the final verdict MUST be 'Failed'.
|
|
c. If there are NO 'Red' or 'Error' statuses, the final verdict is 'Passed'.
|
|
|
|
Your summary should:
|
|
- For a 'Requires Manual Legal Review' verdict, start by stating this. Then, consolidate feedback from all agents, highlighting critical issues ('Red' items) or suggestions ('Amber' items).
|
|
- For an 'Analysis Error' verdict, explain that the proof could not be reliably processed and has been logged for human review. Advise the user to try again with a revised proof.
|
|
- For a 'Failed' status, highlight the critical 'Red' issues that must be addressed.
|
|
- For a 'Passed' status, mention any 'Amber' areas for consideration, if they exist, while maintaining an encouraging tone.
|
|
- Be professional, clear, and constructive.
|
|
|
|
Here are the specialist reviews:
|
|
{self._format_reviews(reviews)}
|
|
|
|
Now, provide your final verdict and summary as a JSON object.
|
|
"""
|
|
|
|
result = await self.gemini.generate_summary(prompt)
|
|
|
|
overall_status = OverallStatus(result.get("overallStatus", "Analysis Error"))
|
|
summary = result.get("summary", "Unable to generate summary.")
|
|
|
|
# Override with financial promotion logic if applicable
|
|
if is_financial_promotion:
|
|
overall_status = OverallStatus.REQUIRES_MANUAL_LEGAL_REVIEW
|
|
|
|
return overall_status, summary, financial_promotion_reason
|
|
|
|
def _format_reviews(self, reviews: dict[str, SubReview]) -> str:
|
|
"""Format reviews as a readable string for the prompt."""
|
|
formatted = []
|
|
for agent_name, review in reviews.items():
|
|
formatted.append(f"""
|
|
{agent_name}:
|
|
RAG Status: {review.ragStatus}
|
|
Feedback: {review.feedback}
|
|
Issues: {review.issues if review.issues else 'None'}
|
|
""")
|
|
return "\n".join(formatted)
|
|
|
|
def determine_status_locally(self, reviews: dict[str, SubReview]) -> OverallStatus:
|
|
"""
|
|
Determine overall status using local logic (without Gemini).
|
|
|
|
This can be used as a fallback or for faster processing.
|
|
"""
|
|
legal_review = reviews.get("Legal Agent")
|
|
|
|
# Check for financial promotion
|
|
if legal_review and legal_review.isFinancialPromotion:
|
|
return OverallStatus.REQUIRES_MANUAL_LEGAL_REVIEW
|
|
|
|
# Check for Error status
|
|
for review in reviews.values():
|
|
if review.ragStatus == RagStatus.ERROR:
|
|
return OverallStatus.ANALYSIS_ERROR
|
|
|
|
# Check for Red status
|
|
for review in reviews.values():
|
|
if review.ragStatus == RagStatus.RED:
|
|
return OverallStatus.FAILED
|
|
|
|
return OverallStatus.PASSED
|