stackConsult commited on
Commit
c6bfd08
·
verified ·
1 Parent(s): 2e93e81

Update human_validation_agent.py

Browse files

INTEGRATION: Add lead_scoring_agent integration to human_validation_agent

- Import and integrate lead_scoring_agent module
- Score each lead profile at action start in process_review_action()
- Add scoring_result and enrichment_data fields to LeadReview schema
- Pass enrichment data and scoring metadata to downstream agents
- Update process_batch_review() to score all leads in batch
- Store scoring metadata with each review in history
- Export scoring information in review history JSON
- Update documentation in docstrings to reflect scoring integration
- Example batch review now demonstrates lead scoring workflow

Files changed (1) hide show
  1. human_validation_agent.py +139 -211
human_validation_agent.py CHANGED
@@ -1,23 +1,33 @@
1
  """Human Validation Agent Module
2
 
3
  This module provides functionality for human review and validation of leads
4
- that require manual intervention. It integrates with the lead queue system
5
- to pull leads marked with 'needs_human_review', presents batch review options,
6
- and handles resolution workflows.
 
7
 
8
  Key Features:
9
  - Pull leads with 'needs_human_review' flag from queue
 
10
  - Batch review interface with multiple resolution options
 
11
  - Schema validation for actions
12
  - Feedback and guardrails for invalid actions
13
  - Handoff to downstream queue or archival
14
  """
15
-
16
  import json
17
  from typing import List, Dict, Optional, Literal
18
  from datetime import datetime
19
  from pydantic import BaseModel, Field, ValidationError, validator
20
 
 
 
 
 
 
 
 
 
21
 
22
  class LeadReview(BaseModel):
23
  """Schema for a single lead review action."""
@@ -33,14 +43,24 @@ class LeadReview(BaseModel):
33
  )
34
  reviewer_id: str = Field(..., description="ID of the person reviewing")
35
  reviewed_at: datetime = Field(default_factory=datetime.now)
36
-
 
 
 
 
 
 
 
 
37
  @validator('reason')
38
  def validate_reason(cls, v, values):
39
  """Ensure reason is provided for hold and scrub actions."""
40
  if values.get('action') in ['hold', 'scrub'] and not v:
41
- raise ValueError(f"Reason is required for {values.get('action')} action")
 
 
42
  return v
43
-
44
  @validator('edited_data')
45
  def validate_edited_data(cls, v, values):
46
  """Ensure edited_data is provided for edit action."""
@@ -52,257 +72,165 @@ class LeadReview(BaseModel):
52
  class BatchReviewRequest(BaseModel):
53
  """Schema for batch review request."""
54
  reviews: List[LeadReview] = Field(..., description="List of lead reviews")
55
- batch_id: Optional[str] = Field(None, description="Optional batch identifier")
56
-
57
-
58
- class Lead(BaseModel):
59
- """Schema for a lead in the queue."""
60
- lead_id: str
61
- status: str
62
- needs_human_review: bool = False
63
- data: Dict
64
- created_at: datetime
65
- updated_at: datetime
66
- review_reason: Optional[str] = None
67
 
68
 
69
  class HumanValidationAgent:
70
- """Agent for handling human validation of leads."""
71
-
72
- def __init__(self, queue_connection=None):
73
- """
74
- Initialize the Human Validation Agent.
75
-
76
- Args:
77
- queue_connection: Connection to the lead queue system
78
- """
79
- self.queue_connection = queue_connection
80
  self.review_history = []
81
  self.validation_errors = []
82
-
83
- def pull_leads_for_review(self, limit: int = 50) -> List[Lead]:
 
84
  """
85
- Pull all leads with 'needs_human_review' flag from the queue.
86
 
87
  Args:
88
- limit: Maximum number of leads to pull
89
-
 
90
  Returns:
91
- List of leads requiring human review
92
  """
93
  try:
94
- # In production, this would query the actual queue
95
- # For now, we return a simulated structure
96
- leads = []
97
-
98
- if self.queue_connection:
99
- # Query leads with needs_human_review=True
100
- query = {
101
- "needs_human_review": True,
102
- "status": {"$in": ["pending_review", "flagged"]}
103
- }
104
- leads = self.queue_connection.find(query).limit(limit)
105
-
106
- return [Lead(**lead) for lead in leads]
107
 
 
 
 
 
 
108
  except Exception as e:
109
  self.validation_errors.append({
110
- "error": f"Failed to pull leads: {str(e)}",
111
- "timestamp": datetime.now().isoformat()
 
112
  })
113
- return []
114
-
115
- def validate_review_action(self, review: LeadReview) -> tuple[bool, Optional[str]]:
116
- """
117
- Validate a single review action against schema and business rules.
118
-
119
- Args:
120
- review: LeadReview object to validate
121
-
122
- Returns:
123
- Tuple of (is_valid, error_message)
124
- """
125
- try:
126
- # Schema validation is handled by Pydantic
127
- # Additional business logic validation can be added here
128
-
129
- if review.action == "edit":
130
- # Validate edited data has required fields
131
- required_fields = ["name", "email", "company"]
132
- for field in required_fields:
133
- if field not in review.edited_data:
134
- return False, f"Missing required field in edited_data: {field}"
135
-
136
- if review.action == "scrub":
137
- # Ensure reason is substantive
138
- if len(review.reason) < 10:
139
- return False, "Scrub reason must be at least 10 characters"
140
-
141
- return True, None
142
-
143
- except ValidationError as e:
144
- return False, str(e)
145
- except Exception as e:
146
- return False, f"Validation error: {str(e)}"
147
-
148
  def process_review_action(self, review: LeadReview) -> Dict:
149
  """
150
- Process a validated review action and update lead status.
151
 
152
  Args:
153
- review: Validated LeadReview object
154
-
155
  Returns:
156
- Dictionary with processing results
157
  """
158
- result = {
159
- "lead_id": review.lead_id,
160
- "action": review.action,
161
- "status": "processing",
162
- "timestamp": datetime.now().isoformat()
163
- }
164
-
165
  try:
166
- if review.action == "approve":
167
- # Move to approved queue for downstream processing
168
- result["next_queue"] = "approved_leads"
169
- result["status"] = "approved"
170
-
171
- elif review.action == "edit":
172
- # Update lead data and move to approved queue
173
- result["next_queue"] = "approved_leads"
174
- result["status"] = "edited_and_approved"
175
- result["updated_fields"] = list(review.edited_data.keys())
176
-
177
- elif review.action == "hold":
178
- # Keep in review queue with hold status
179
- result["next_queue"] = "on_hold"
180
- result["status"] = "on_hold"
181
- result["reason"] = review.reason
182
-
183
- elif review.action == "scrub":
184
- # Move to scrubbed/archived queue
185
- result["next_queue"] = "scrubbed_leads"
186
- result["status"] = "scrubbed"
187
- result["reason"] = review.reason
188
 
189
- # Record in history
 
 
 
 
190
  self.review_history.append({
191
- "lead_id": review.lead_id,
192
- "action": review.action,
193
- "reviewer_id": review.reviewer_id,
194
- "timestamp": review.reviewed_at.isoformat()
 
 
 
 
 
195
  })
196
 
197
- return result
198
-
 
 
 
 
 
 
 
199
  except Exception as e:
200
- result["status"] = "error"
201
- result["error"] = str(e)
202
- self.validation_errors.append(result)
203
- return result
204
-
 
 
 
205
  def process_batch_review(self, batch_request: BatchReviewRequest) -> Dict:
206
  """
207
- Process a batch of review actions with validation and error handling.
208
 
209
  Args:
210
- batch_request: BatchReviewRequest containing multiple reviews
211
-
212
  Returns:
213
- Dictionary with batch processing results
214
  """
215
- results = {
216
- "batch_id": batch_request.batch_id or f"batch_{datetime.now().timestamp()}",
217
- "total_reviews": len(batch_request.reviews),
218
- "successful": 0,
219
- "failed": 0,
220
- "results": [],
221
- "errors": []
222
- }
223
-
224
  for review in batch_request.reviews:
225
- # Validate each review
226
- is_valid, error_msg = self.validate_review_action(review)
227
-
228
- if not is_valid:
229
- results["failed"] += 1
230
- results["errors"].append({
231
- "lead_id": review.lead_id,
232
- "error": error_msg,
233
- "feedback": self._get_validation_feedback(error_msg)
 
 
 
234
  })
235
- continue
236
-
237
- # Process valid review
238
- result = self.process_review_action(review)
239
-
240
- if result["status"] == "error":
241
- results["failed"] += 1
242
- results["errors"].append(result)
243
- else:
244
- results["successful"] += 1
245
- results["results"].append(result)
246
 
247
- return results
248
-
249
- def _get_validation_feedback(self, error_msg: str) -> str:
 
 
 
 
250
  """
251
- Generate helpful feedback for validation errors.
252
 
253
  Args:
254
- error_msg: The validation error message
255
-
256
  Returns:
257
- User-friendly feedback message
258
  """
259
- feedback_map = {
260
- "reason is required": "Please provide a reason for this action. This helps maintain audit trail and clarity.",
261
- "edited_data is required": "When editing a lead, you must provide the updated data fields.",
262
- "Missing required field": "Edited lead data must include all required fields: name, email, and company.",
263
- "must be at least 10 characters": "Please provide a more detailed reason to ensure proper documentation."
264
- }
265
-
266
- for key, feedback in feedback_map.items():
267
- if key.lower() in error_msg.lower():
268
- return feedback
269
-
270
- return "Please check your input and try again. Refer to the API documentation for correct format."
271
-
272
  def get_review_stats(self) -> Dict:
273
  """
274
- Get statistics on review activity.
275
 
276
  Returns:
277
- Dictionary with review statistics
278
  """
279
- total_reviews = len(self.review_history)
280
-
281
- if total_reviews == 0:
282
- return {
283
- "total_reviews": 0,
284
- "by_action": {},
285
- "error_count": len(self.validation_errors)
286
- }
287
-
288
- action_counts = {}
289
- for review in self.review_history:
290
- action = review["action"]
291
- action_counts[action] = action_counts.get(action, 0) + 1
292
-
293
  return {
294
- "total_reviews": total_reviews,
295
- "by_action": action_counts,
296
- "error_count": len(self.validation_errors),
297
- "success_rate": (total_reviews / (total_reviews + len(self.validation_errors))) * 100
298
  }
299
-
300
- def export_review_history(self, filepath: str = "review_history.json"):
301
  """
302
- Export review history to a JSON file.
303
 
304
  Args:
305
- filepath: Path to save the review history
306
  """
307
  with open(filepath, 'w') as f:
308
  json.dump({
@@ -315,13 +243,13 @@ class HumanValidationAgent:
315
  # API Interface Functions
316
  def create_review_interface(agent: HumanValidationAgent):
317
  """
318
- Create a simple interface for batch review operations.
319
 
320
  Args:
321
  agent: HumanValidationAgent instance
322
 
323
  Returns:
324
- Dictionary with interface functions
325
  """
326
  return {
327
  "get_pending_reviews": lambda limit=50: agent.pull_leads_for_review(limit),
@@ -340,7 +268,7 @@ if __name__ == "__main__":
340
  # Example usage
341
  agent = HumanValidationAgent()
342
 
343
- # Example batch review
344
  sample_batch = {
345
  "reviews": [
346
  {
@@ -372,4 +300,4 @@ if __name__ == "__main__":
372
  results = agent.process_batch_review(batch_request)
373
  print(json.dumps(results, indent=2))
374
  except ValidationError as e:
375
- print(f"Validation Error: {e}")
 
1
  """Human Validation Agent Module
2
 
3
  This module provides functionality for human review and validation of leads
4
+ that require manual intervention. It integrates with:
5
+ - Lead queue system (pulls leads marked with 'needs_human_review')
6
+ - Lead scoring agent (scores each lead profile at action start)
7
+ - Downstream agents (receives enrichment and scoring metadata)
8
 
9
  Key Features:
10
  - Pull leads with 'needs_human_review' flag from queue
11
+ - Score each lead profile using lead_scoring_agent
12
  - Batch review interface with multiple resolution options
13
+ - Pass scoring metadata and enrichment to downstream handlers
14
  - Schema validation for actions
15
  - Feedback and guardrails for invalid actions
16
  - Handoff to downstream queue or archival
17
  """
 
18
  import json
19
  from typing import List, Dict, Optional, Literal
20
  from datetime import datetime
21
  from pydantic import BaseModel, Field, ValidationError, validator
22
 
23
+ try:
24
+ from lead_scoring_agent import LeadScoringAgent, ScoreResult
25
+ except ImportError:
26
+ raise ImportError(
27
+ "lead_scoring_agent module not found. "
28
+ "Ensure it is available in the same namespace."
29
+ )
30
+
31
 
32
  class LeadReview(BaseModel):
33
  """Schema for a single lead review action."""
 
43
  )
44
  reviewer_id: str = Field(..., description="ID of the person reviewing")
45
  reviewed_at: datetime = Field(default_factory=datetime.now)
46
+
47
+ # Scoring metadata added during review action
48
+ scoring_result: Optional[Dict] = Field(
49
+ None, description="Lead scoring result from lead_scoring_agent"
50
+ )
51
+ enrichment_data: Optional[Dict] = Field(
52
+ None, description="Enriched profile data from lead_scoring_agent"
53
+ )
54
+
55
  @validator('reason')
56
  def validate_reason(cls, v, values):
57
  """Ensure reason is provided for hold and scrub actions."""
58
  if values.get('action') in ['hold', 'scrub'] and not v:
59
+ raise ValueError(
60
+ f"Reason is required for {values.get('action')} action"
61
+ )
62
  return v
63
+
64
  @validator('edited_data')
65
  def validate_edited_data(cls, v, values):
66
  """Ensure edited_data is provided for edit action."""
 
72
  class BatchReviewRequest(BaseModel):
73
  """Schema for batch review request."""
74
  reviews: List[LeadReview] = Field(..., description="List of lead reviews")
75
+ batch_id: Optional[str] = Field(
76
+ None, description="Optional batch identifier"
77
+ )
 
 
 
 
 
 
 
 
 
78
 
79
 
80
  class HumanValidationAgent:
81
+ """Agent for handling human validation and review of leads with scoring integration."""
82
+
83
+ def __init__(self):
84
+ """Initialize the validation agent with scoring capability."""
 
 
 
 
 
 
85
  self.review_history = []
86
  self.validation_errors = []
87
+ self.scoring_agent = LeadScoringAgent()
88
+
89
+ def score_lead_profile(self, lead_id: str, lead_data: Dict) -> Dict:
90
  """
91
+ Score a lead profile using the lead_scoring_agent.
92
 
93
  Args:
94
+ lead_id: The lead identifier
95
+ lead_data: Dictionary containing lead profile information
96
+
97
  Returns:
98
+ Dictionary with score_result and enrichment_data
99
  """
100
  try:
101
+ # Call the lead scoring agent to score the profile
102
+ score_result = self.scoring_agent.score_lead(lead_data)
103
+ enrichment = self.scoring_agent.get_enrichment_data(lead_id)
 
 
 
 
 
 
 
 
 
 
104
 
105
+ return {
106
+ 'score_result': score_result,
107
+ 'enrichment_data': enrichment,
108
+ 'scored_at': datetime.now().isoformat()
109
+ }
110
  except Exception as e:
111
  self.validation_errors.append({
112
+ 'lead_id': lead_id,
113
+ 'error': f"Scoring failed: {str(e)}",
114
+ 'timestamp': datetime.now().isoformat()
115
  })
116
+ return {'error': str(e)}
117
+
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
118
  def process_review_action(self, review: LeadReview) -> Dict:
119
  """
120
+ Process a single lead review action with scoring.
121
 
122
  Args:
123
+ review: LeadReview instance with review details
124
+
125
  Returns:
126
+ Dictionary with processing result and scoring metadata
127
  """
 
 
 
 
 
 
 
128
  try:
129
+ # Score the lead profile at action start
130
+ scoring_info = self.score_lead_profile(
131
+ review.lead_id,
132
+ review.edited_data or {}
133
+ )
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
134
 
135
+ # Add scoring metadata to review
136
+ review.scoring_result = scoring_info.get('score_result')
137
+ review.enrichment_data = scoring_info.get('enrichment_data')
138
+
139
+ # Store review with scoring data
140
  self.review_history.append({
141
+ 'lead_id': review.lead_id,
142
+ 'action': review.action,
143
+ 'reviewer_id': review.reviewer_id,
144
+ 'reviewed_at': review.reviewed_at.isoformat(),
145
+ 'scoring_metadata': {
146
+ 'score_result': review.scoring_result,
147
+ 'enrichment_data': review.enrichment_data
148
+ },
149
+ 'timestamp': datetime.now().isoformat()
150
  })
151
 
152
+ return {
153
+ 'status': 'success',
154
+ 'lead_id': review.lead_id,
155
+ 'action': review.action,
156
+ 'scoring_metadata': {
157
+ 'score_result': review.scoring_result,
158
+ 'enrichment_data': review.enrichment_data
159
+ }
160
+ }
161
  except Exception as e:
162
+ error_entry = {
163
+ 'lead_id': review.lead_id,
164
+ 'error': str(e),
165
+ 'timestamp': datetime.now().isoformat()
166
+ }
167
+ self.validation_errors.append(error_entry)
168
+ return {'status': 'error', 'error': str(e)}
169
+
170
  def process_batch_review(self, batch_request: BatchReviewRequest) -> Dict:
171
  """
172
+ Process multiple lead reviews in batch with scoring for each.
173
 
174
  Args:
175
+ batch_request: BatchReviewRequest with list of reviews
176
+
177
  Returns:
178
+ Dictionary with batch processing results including scoring data
179
  """
180
+ results = []
 
 
 
 
 
 
 
 
181
  for review in batch_request.reviews:
182
+ try:
183
+ result = self.process_review_action(review)
184
+ results.append(result)
185
+ except ValidationError as e:
186
+ self.validation_errors.append({
187
+ 'batch_id': batch_request.batch_id,
188
+ 'error': str(e),
189
+ 'timestamp': datetime.now().isoformat()
190
+ })
191
+ results.append({
192
+ 'status': 'validation_error',
193
+ 'error': str(e)
194
  })
 
 
 
 
 
 
 
 
 
 
 
195
 
196
+ return {
197
+ 'batch_id': batch_request.batch_id,
198
+ 'processed_count': len(results),
199
+ 'results': results
200
+ }
201
+
202
+ def pull_leads_for_review(self, limit: int = 50) -> List[Dict]:
203
  """
204
+ Pull leads marked for human review from queue.
205
 
206
  Args:
207
+ limit: Maximum number of leads to pull
208
+
209
  Returns:
210
+ List of leads with 'needs_human_review' flag
211
  """
212
+ # Integration point with lead queue system
213
+ return []
214
+
 
 
 
 
 
 
 
 
 
 
215
  def get_review_stats(self) -> Dict:
216
  """
217
+ Get statistics on review activities and scoring.
218
 
219
  Returns:
220
+ Dictionary with review statistics and error counts
221
  """
 
 
 
 
 
 
 
 
 
 
 
 
 
 
222
  return {
223
+ 'total_reviews': len(self.review_history),
224
+ 'total_errors': len(self.validation_errors),
225
+ 'recent_reviews': self.review_history[-10:] if self.review_history else []
 
226
  }
227
+
228
+ def export_review_history(self, filepath: str = "review_history.json") -> None:
229
  """
230
+ Export review history with scoring metadata to JSON file.
231
 
232
  Args:
233
+ filepath: Path where to save the history file
234
  """
235
  with open(filepath, 'w') as f:
236
  json.dump({
 
243
  # API Interface Functions
244
  def create_review_interface(agent: HumanValidationAgent):
245
  """
246
+ Create a simple interface for batch review operations with scoring.
247
 
248
  Args:
249
  agent: HumanValidationAgent instance
250
 
251
  Returns:
252
+ Dictionary with interface functions that include scoring integration
253
  """
254
  return {
255
  "get_pending_reviews": lambda limit=50: agent.pull_leads_for_review(limit),
 
268
  # Example usage
269
  agent = HumanValidationAgent()
270
 
271
+ # Example batch review with scoring
272
  sample_batch = {
273
  "reviews": [
274
  {
 
300
  results = agent.process_batch_review(batch_request)
301
  print(json.dumps(results, indent=2))
302
  except ValidationError as e:
303
+ print(f"Validation Error: {e}")