Last active
July 19, 2025 07:54
-
-
Save muthuspark/2cad1c303a4f145e72edb75ec9ab9cf7 to your computer and use it in GitHub Desktop.
Accuracy Refinement MultiAgent Crew
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import json | |
| import re | |
| from crewai import Agent, Task, Crew, Process | |
| from .llms import gemini_llm, ollama_llm | |
| # Define agents | |
| validation_filter_agent = Agent( | |
| role="Validation Requirement Assessor", | |
| goal="Determine if content requires validation based on complexity and factual claims.", | |
| backstory="You quickly assess whether text needs validation by identifying simple greetings vs factual content.", | |
| llm=ollama_llm, | |
| allow_delegation=False, | |
| verbose=False | |
| ) | |
| accuracy_validator_agent = Agent( | |
| role="Accuracy Validation Specialist", | |
| goal="Evaluate factual correctness and refine content when needed.", | |
| backstory="You are a fact-checker who identifies inaccuracies and improves content quality.", | |
| llm=gemini_llm, | |
| allow_delegation=False, | |
| verbose=False | |
| ) | |
| readability_optimizer_agent = Agent( | |
| role="Readability Optimization Expert", | |
| goal="Optimize content for clarity while maintaining accuracy.", | |
| backstory="You simplify complex content to improve readability without losing meaning.", | |
| llm=gemini_llm, | |
| allow_delegation=False, | |
| verbose=False | |
| ) | |
| content_enhancer_agent = Agent( | |
| role="Content Enhancement Specialist", | |
| goal="Apply final polish for grammar, style, and formatting.", | |
| backstory="You are an editor who perfects content presentation and quality.", | |
| llm=gemini_llm, | |
| allow_delegation=False, | |
| verbose=False | |
| ) | |
| def process_text_with_crew(query: str, llm_response: str, accuracy_threshold: int = 9, | |
| readability_threshold: int = 70, enable_enhancement: bool = True, debug: bool = False) -> str: | |
| """Process text through validation pipeline using CrewAI.""" | |
| if debug: | |
| print("[DEBUG] Starting validation filter agent...") | |
| print(f"[DEBUG] Query: {query}") | |
| print(f"[DEBUG] LLM Response: {llm_response}") | |
| # Check if validation is needed | |
| validation_task = Task( | |
| description=f"""Determine if this response needs validation: | |
| User Query with History: {query} | |
| LLM Response: {llm_response} | |
| Skip validation for: greetings, acknowledgments, simple opinions | |
| Validate: factual claims, technical explanations, complex reasoning""", | |
| expected_output='{"validation_required": "YES/NO", "reasoning": "brief explanation"}', | |
| agent=validation_filter_agent | |
| ) | |
| filter_crew = Crew( | |
| agents=[validation_filter_agent], | |
| tasks=[validation_task], | |
| process=Process.sequential | |
| ) | |
| filter_result = filter_crew.kickoff() | |
| if debug: | |
| print(f"[DEBUG] Validation filter result: {filter_result}") | |
| if _extract_json_field(str(filter_result), "validation_required") != "YES": | |
| if debug: | |
| print("[DEBUG] Validation not required, returning original response.") | |
| return llm_response | |
| # Define validation tasks | |
| tasks = [ | |
| Task( | |
| description=f"""Evaluate accuracy (1-10 scale) and refine if score < {accuracy_threshold}: | |
| Query: {query} | |
| Response: {llm_response}""", | |
| expected_output='{"accuracy_score": 1-10, "refined_content": "improved version or APPROVED"}', | |
| agent=accuracy_validator_agent | |
| ), | |
| Task( | |
| description=f"""Optimize the given text for readability (target Flesch score >= {readability_threshold}): | |
| - Simplify sentences (15-20 words) | |
| - Use clear vocabulary | |
| - Improve flow""", | |
| expected_output='{"optimized_content": "restructured version or APPROVED"}', | |
| agent=readability_optimizer_agent | |
| ) | |
| ] | |
| agents = [accuracy_validator_agent, readability_optimizer_agent] | |
| if enable_enhancement: | |
| tasks.append(Task( | |
| description="Apply final polish: grammar, style, formatting", | |
| expected_output='{"final_content": "polished version"}', | |
| agent=content_enhancer_agent | |
| )) | |
| agents.append(content_enhancer_agent) | |
| if debug: | |
| print(f"[DEBUG] Validation pipeline agents: {[agent.role for agent in agents]}") | |
| print(f"[DEBUG] Validation pipeline tasks: {[task.description for task in tasks]}") | |
| # Execute validation pipeline | |
| validation_crew = Crew( | |
| agents=agents, | |
| tasks=tasks, | |
| process=Process.sequential | |
| ) | |
| result = validation_crew.kickoff() | |
| if debug: | |
| print(f"[DEBUG] Validation pipeline result: {result}") | |
| final_content = _extract_content(str(result)) or llm_response | |
| if debug: | |
| print(f"[DEBUG] Final content: {final_content}") | |
| return final_content | |
| def _extract_json_field(text: str, field: str) -> str: | |
| """Extract field from JSON in text.""" | |
| try: | |
| json_match = re.search(r'\{.*\}', text, re.DOTALL) | |
| if json_match: | |
| return json.loads(json_match.group()).get(field, "") | |
| except json.JSONDecodeError: | |
| pass | |
| return "" | |
| def _extract_content(result: str) -> str: | |
| """Extract the best available content from crew result.""" | |
| # Priority: final_content > optimized_content > refined_content | |
| for field in ["final_content", "optimized_content", "refined_content"]: | |
| content = _extract_json_field(result, field) | |
| if content and content.strip() and content != "APPROVED": | |
| return content.strip() | |
| return "" | |
| def process_text_with_agent(query: str, llm_response: str) -> str: | |
| """Convenience function with default settings.""" | |
| return process_text_with_crew(query, llm_response) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment