File size: 8,402 Bytes
f844f16 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 |
"""
Answer Formatter - Final answer formatting according to GAIA requirements
The Answer Formatter is responsible for:
1. Taking the draft answer and formatting it according to GAIA rules
2. Extracting the final answer from comprehensive responses
3. Ensuring exact-match compliance
4. Handling different answer types (numbers, strings, lists)
"""
import re
from typing import Dict, Any
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langgraph.types import Command
from langchain_groq import ChatGroq
from observability import agent_span
from dotenv import load_dotenv
load_dotenv("env.local")
def extract_final_answer(text: str) -> str:
"""
Extract the final answer from text following GAIA formatting rules.
GAIA Rules:
• Single number → write the number only (no commas, units, or other symbols)
• Single string/phrase → write the text only; omit articles and abbreviations unless explicitly required
• List → separate elements with a single comma and a space
• Never include surrounding text, quotes, brackets, or markdown
"""
if not text or not text.strip():
return ""
# Clean the text
text = text.strip()
# Look for explicit final answer markers
answer_patterns = [
r"final answer[:\s]*(.+?)(?:\n|$)",
r"answer[:\s]*(.+?)(?:\n|$)",
r"result[:\s]*(.+?)(?:\n|$)",
r"conclusion[:\s]*(.+?)(?:\n|$)"
]
for pattern in answer_patterns:
match = re.search(pattern, text, re.IGNORECASE | re.MULTILINE)
if match:
text = match.group(1).strip()
break
# Remove common prefixes/suffixes
prefixes_to_remove = [
"the answer is", "it is", "this is", "that is",
"final answer:", "answer:", "result:", "conclusion:",
"therefore", "thus", "so", "hence"
]
for prefix in prefixes_to_remove:
if text.lower().startswith(prefix.lower()):
text = text[len(prefix):].strip()
# Remove quotes, brackets, and markdown
text = re.sub(r'^["\'\[\(]|["\'\]\)]$', '', text)
text = re.sub(r'^\*\*|\*\*$', '', text) # Remove bold markdown
text = re.sub(r'^`|`$', '', text) # Remove code markdown
# Handle different answer types
# Check if it's a pure number
number_match = re.match(r'^-?\d+(?:\.\d+)?$', text.strip())
if number_match:
# Return number without formatting
num = float(text.strip()) if '.' in text else int(text.strip())
return str(int(num)) if num == int(num) else str(num)
# Check if it's a list (comma-separated)
if ',' in text:
items = [item.strip() for item in text.split(',')]
# Clean each item
cleaned_items = []
for item in items:
item = re.sub(r'^["\'\[\(]|["\'\]\)]$', '', item.strip())
if item:
cleaned_items.append(item)
return ', '.join(cleaned_items)
# For single strings, remove articles if they're not essential
# But be careful not to remove essential parts
words = text.split()
if len(words) > 1 and words[0].lower() in ['the', 'a', 'an']:
# Only remove if the rest makes sense
remaining = ' '.join(words[1:])
if remaining and len(remaining) > 2:
text = remaining
return text.strip()
def load_formatter_prompt() -> str:
"""Load the formatting prompt"""
try:
with open("archive/prompts/verification_prompt.txt", "r") as f:
return f.read()
except FileNotFoundError:
return """
You are a final answer formatter ensuring compliance with GAIA benchmark requirements.
Your task is to extract the precise final answer from a comprehensive response.
CRITICAL FORMATTING RULES:
• Single number → write the number only (no commas, units, or symbols)
• Single string/phrase → write the text only; omit articles unless required
• List → separate elements with comma and space
• NEVER include surrounding text like "Final Answer:", quotes, brackets, or markdown
• The response must contain ONLY the answer itself
Examples:
Question: "What is 25 + 17?"
Draft: "After calculating, the answer is 42."
Formatted: "42"
Question: "What is the capital of France?"
Draft: "The capital of France is Paris."
Formatted: "Paris"
Question: "List the first 3 prime numbers"
Draft: "The first three prime numbers are 2, 3, and 5."
Formatted: "2, 3, 5"
Extract ONLY the final answer following these rules exactly.
"""
def answer_formatter(state: Dict[str, Any]) -> Command:
"""
Answer Formatter node that creates GAIA-compliant final answers.
Takes the draft_answer and formats it according to GAIA requirements.
Returns Command to END the workflow.
"""
print("📝 Answer Formatter: Creating final formatted answer...")
try:
# Get formatting prompt
formatter_prompt = load_formatter_prompt()
# Initialize LLM for formatting
llm = ChatGroq(
model="llama-3.3-70b-versatile",
temperature=0.0, # Zero temperature for consistent formatting
max_tokens=512
)
# Create agent span for tracing
with agent_span(
"formatter",
metadata={
"draft_answer_length": len(state.get("draft_answer", "")),
"user_id": state.get("user_id", "unknown"),
"session_id": state.get("session_id", "unknown")
}
) as span:
# Get the draft answer
draft_answer = state.get("draft_answer", "")
if not draft_answer:
final_answer = "No answer could be generated."
else:
# Get the original question for context
messages = state.get("messages", [])
user_query = ""
for msg in messages:
if isinstance(msg, HumanMessage):
user_query = msg.content
break
# Build formatting request
formatting_request = f"""
Extract the final answer from this comprehensive response following GAIA formatting rules:
Original Question: {user_query}
Draft Response:
{draft_answer}
Instructions:
1. Identify the core answer within the draft response
2. Remove all explanatory text, prefixes, and formatting
3. Apply GAIA formatting rules exactly
4. Return ONLY the final answer
What is the properly formatted final answer?
"""
# Create messages for formatting
formatting_messages = [
SystemMessage(content=formatter_prompt),
HumanMessage(content=formatting_request)
]
# Get formatted response
response = llm.invoke(formatting_messages)
# Extract the final answer using our utility function
final_answer = extract_final_answer(response.content)
# Fallback: if extraction fails, try direct extraction from draft
if not final_answer or len(final_answer) < 1:
print("⚠️ LLM formatting failed, using direct extraction")
final_answer = extract_final_answer(draft_answer)
# Final fallback
if not final_answer:
final_answer = "Unable to extract a clear answer."
print(f"📝 Answer Formatter: Final answer = '{final_answer}'")
# Update trace
if span:
span.update_trace(output={"final_answer": final_answer})
# Return command to END the workflow
return Command(
goto="__end__",
update={
"final_answer": final_answer
}
)
except Exception as e:
print(f"❌ Answer Formatter Error: {e}")
# Return error as final answer
return Command(
goto="__end__",
update={
"final_answer": f"Error formatting answer: {str(e)}"
}
) |