a957cf0276
Previously, proof metadata collected during upload was only used for database persistence. Now it flows through the entire analysis pipeline so agents can tailor their feedback to the specific channel and format being reviewed. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
201 lines
8 KiB
Python
Executable file
201 lines
8 KiB
Python
Executable file
from typing import Optional
|
|
|
|
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
|
|
|
|
def _build_revision_context(self, previous_analysis: dict, reviews: dict[str, SubReview]) -> str:
|
|
"""Build revision context section for the synthesis prompt."""
|
|
previous_version = previous_analysis.get("version", "N/A")
|
|
|
|
# Aggregate resolved/outstanding/new issues across all agents
|
|
all_resolved = []
|
|
all_outstanding = []
|
|
all_new = []
|
|
|
|
for agent_name, review in reviews.items():
|
|
if review.resolvedIssues:
|
|
all_resolved.extend([f"[{agent_name}] {issue}" for issue in review.resolvedIssues])
|
|
if review.outstandingIssues:
|
|
all_outstanding.extend([f"[{agent_name}] {issue}" for issue in review.outstandingIssues])
|
|
if review.newIssues:
|
|
all_new.extend([f"[{agent_name}] {issue}" for issue in review.newIssues])
|
|
|
|
resolved_list = "\n".join(f" - {issue}" for issue in all_resolved) if all_resolved else " (None)"
|
|
outstanding_list = "\n".join(f" - {issue}" for issue in all_outstanding) if all_outstanding else " (None)"
|
|
new_list = "\n".join(f" - {issue}" for issue in all_new) if all_new else " (None)"
|
|
|
|
return f"""
|
|
---
|
|
|
|
**REVISION CONTEXT**
|
|
|
|
This is a revision (comparing against Version {previous_version}). The analysis has identified the following changes:
|
|
|
|
**RESOLVED ISSUES** (fixed in this revision):
|
|
{resolved_list}
|
|
|
|
**OUTSTANDING ISSUES** (still need attention):
|
|
{outstanding_list}
|
|
|
|
**NEW ISSUES** (introduced in this revision):
|
|
{new_list}
|
|
|
|
In your summary:
|
|
1. Acknowledge this is a revision and summarize the progress made
|
|
2. Highlight any resolved issues positively
|
|
3. Emphasize outstanding issues that still need to be addressed
|
|
4. Call out any regressions (new issues) that were introduced
|
|
|
|
---
|
|
"""
|
|
|
|
async def synthesize(
|
|
self,
|
|
reviews: dict[str, SubReview],
|
|
previous_analysis: Optional[dict] = None,
|
|
channel: Optional[str] = None,
|
|
sub_channel: Optional[str] = None,
|
|
proof_type: Optional[str] = None,
|
|
) -> tuple[OverallStatus, str, str | None]:
|
|
"""
|
|
Synthesize specialist reviews into final verdict and summary.
|
|
|
|
Args:
|
|
reviews: Dictionary mapping agent names to their SubReview results
|
|
previous_analysis: Optional dict containing the previous version's analysis
|
|
results. When provided, enables revision-aware summary.
|
|
channel: Target channel (e.g. "Social", "Digital")
|
|
sub_channel: Target sub-channel (e.g. "Meta", "Google")
|
|
proof_type: Proof format type (e.g. "In-feed 1x1", "Banner")
|
|
|
|
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 revision context if previous analysis is available
|
|
revision_context = ""
|
|
if previous_analysis and previous_analysis.get("version"):
|
|
revision_context = self._build_revision_context(previous_analysis, reviews)
|
|
|
|
# Build proof metadata context
|
|
metadata_context = f"""
|
|
**PROOF METADATA**
|
|
- Channel: {channel or "Not specified"}
|
|
- Sub-Channel: {sub_channel or "Not specified"}
|
|
- Proof Type: {proof_type or "Not specified"}
|
|
"""
|
|
|
|
# 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.
|
|
|
|
{metadata_context}
|
|
|
|
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.
|
|
|
|
**Response Format:**
|
|
- Start with a one-line verdict statement
|
|
- List key issues as bullet points (max 3-5 bullets)
|
|
- Each bullet: one sentence, actionable
|
|
- For 'Passed': briefly note any amber items in 1-2 bullets
|
|
{revision_context}
|
|
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
|