File size: 14,712 Bytes
66dbebd |
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 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 |
"""
Response Synthesis Agent
Specialized in integrating multiple agent outputs into coherent responses
"""
import logging
from typing import Dict, Any, List
import re
logger = logging.getLogger(__name__)
class ResponseSynthesisAgent:
def __init__(self, llm_router=None):
self.llm_router = llm_router
self.agent_id = "RESP_SYNTH_001"
self.specialization = "Multi-source information integration and coherent response generation"
# Response templates for different intent types
self.response_templates = {
"information_request": {
"structure": "introduction → key_points → conclusion",
"tone": "informative, clear, authoritative"
},
"task_execution": {
"structure": "confirmation → steps → expected_outcome",
"tone": "action-oriented, precise, reassuring"
},
"creative_generation": {
"structure": "concept → development → refinement",
"tone": "creative, engaging, expressive"
},
"analysis_research": {
"structure": "hypothesis → analysis → insights",
"tone": "analytical, evidence-based, objective"
},
"casual_conversation": {
"structure": "engagement → response → follow_up",
"tone": "friendly, conversational, natural"
}
}
async def execute(self, agent_outputs: List[Dict[str, Any]], user_input: str,
context: Dict[str, Any] = None, **kwargs) -> Dict[str, Any]:
"""
Synthesize responses from multiple agent outputs
"""
try:
logger.info(f"{self.agent_id} synthesizing {len(agent_outputs)} agent outputs")
# Extract intent information
intent_info = self._extract_intent_info(agent_outputs)
primary_intent = intent_info.get('primary_intent', 'casual_conversation')
# Structure the synthesis process
synthesis_result = await self._synthesize_response(
agent_outputs, user_input, context, primary_intent
)
# Add quality metrics
synthesis_result.update({
"agent_id": self.agent_id,
"synthesis_quality_metrics": self._calculate_quality_metrics(synthesis_result),
"intent_alignment": self._check_intent_alignment(synthesis_result, intent_info)
})
logger.info(f"{self.agent_id} completed synthesis")
return synthesis_result
except Exception as e:
logger.error(f"{self.agent_id} synthesis error: {str(e)}")
return self._get_fallback_response(user_input, agent_outputs)
async def _synthesize_response(self, agent_outputs: List[Dict[str, Any]],
user_input: str, context: Dict[str, Any],
primary_intent: str) -> Dict[str, Any]:
"""Synthesize responses using appropriate method based on intent"""
if self.llm_router:
# Use LLM for sophisticated synthesis
return await self._llm_based_synthesis(agent_outputs, user_input, context, primary_intent)
else:
# Use template-based synthesis
return await self._template_based_synthesis(agent_outputs, user_input, primary_intent)
async def _llm_based_synthesis(self, agent_outputs: List[Dict[str, Any]],
user_input: str, context: Dict[str, Any],
primary_intent: str) -> Dict[str, Any]:
"""Use LLM for sophisticated response synthesis"""
synthesis_prompt = self._build_synthesis_prompt(agent_outputs, user_input, context, primary_intent)
# Simulate LLM synthesis (replace with actual LLM call)
synthesized_response = await self._template_based_synthesis(agent_outputs, user_input, primary_intent)
# Enhance with simulated LLM improvements
draft_response = synthesized_response["final_response"]
enhanced_response = self._enhance_response_quality(draft_response, primary_intent)
return {
"draft_response": draft_response,
"final_response": enhanced_response,
"source_references": self._extract_source_references(agent_outputs),
"coherence_score": 0.85,
"improvement_opportunities": self._identify_improvements(enhanced_response),
"synthesis_method": "llm_enhanced"
}
async def _template_based_synthesis(self, agent_outputs: List[Dict[str, Any]],
user_input: str, primary_intent: str) -> Dict[str, Any]:
"""Template-based response synthesis"""
template = self.response_templates.get(primary_intent, self.response_templates["casual_conversation"])
# Extract relevant content from agent outputs
content_blocks = self._extract_content_blocks(agent_outputs)
# Apply template structure
structured_response = self._apply_response_template(content_blocks, template, primary_intent)
return {
"draft_response": structured_response,
"final_response": structured_response, # No enhancement in template mode
"source_references": self._extract_source_references(agent_outputs),
"coherence_score": 0.75,
"improvement_opportunities": ["Consider adding more specific details"],
"synthesis_method": "template_based"
}
def _build_synthesis_prompt(self, agent_outputs: List[Dict[str, Any]],
user_input: str, context: Dict[str, Any],
primary_intent: str) -> str:
"""Build prompt for LLM-based synthesis"""
return f"""
Synthesize a coherent response from multiple AI agent outputs:
User Question: "{user_input}"
Primary Intent: {primary_intent}
Agent Outputs to Integrate:
{self._format_agent_outputs_for_synthesis(agent_outputs)}
Conversation Context: {context.get('conversation_history', [])[-3:] if context else 'No context'}
Requirements:
- Maintain accuracy from source materials
- Ensure logical flow and coherence
- Match the {primary_intent} intent style
- Keep response concise but comprehensive
- Include relevant details from agent outputs
Provide a well-structured, natural-sounding response.
"""
def _extract_intent_info(self, agent_outputs: List[Dict[str, Any]]) -> Dict[str, Any]:
"""Extract intent information from agent outputs"""
for output in agent_outputs:
if 'primary_intent' in output:
return {
'primary_intent': output['primary_intent'],
'confidence': output.get('confidence_scores', {}).get(output['primary_intent'], 0.5),
'source_agent': output.get('agent_id', 'unknown')
}
return {'primary_intent': 'casual_conversation', 'confidence': 0.5}
def _extract_content_blocks(self, agent_outputs: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Extract content blocks from agent outputs for synthesis"""
content_blocks = []
for output in agent_outputs:
if 'result' in output:
content_blocks.append({
'content': output['result'],
'source': output.get('agent_id', 'unknown'),
'confidence': output.get('confidence', 0.5)
})
elif 'primary_intent' in output:
content_blocks.append({
'content': f"Intent analysis: {output['primary_intent']}",
'source': output.get('agent_id', 'intent_agent'),
'confidence': output.get('confidence_scores', {}).get(output['primary_intent'], 0.5)
})
elif 'final_response' in output:
content_blocks.append({
'content': output['final_response'],
'source': output.get('agent_id', 'unknown'),
'confidence': output.get('confidence_score', 0.7)
})
return content_blocks
def _apply_response_template(self, content_blocks: List[Dict[str, Any]],
template: Dict[str, str], intent: str) -> str:
"""Apply response template to structure the content"""
if intent == "information_request":
return self._structure_informative_response(content_blocks)
elif intent == "task_execution":
return self._structure_actionable_response(content_blocks)
else:
return self._structure_conversational_response(content_blocks)
def _structure_informative_response(self, content_blocks: List[Dict[str, Any]]) -> str:
"""Structure an informative response (intro → key_points → conclusion)"""
if not content_blocks:
return "I'm here to help! Could you provide more details about what you're looking for?"
intro = f"Based on the information available"
key_points = "\n".join([f"• {block['content']}" for block in content_blocks[:3]])
conclusion = "I hope this helps! Let me know if you need any clarification."
return f"{intro}:\n\n{key_points}\n\n{conclusion}"
def _structure_actionable_response(self, content_blocks: List[Dict[str, Any]]) -> str:
"""Structure an actionable response (confirmation → steps → outcome)"""
if not content_blocks:
return "I understand you'd like some help. What specific task would you like to accomplish?"
confirmation = "I can help with that!"
steps = "\n".join([f"{i+1}. {block['content']}" for i, block in enumerate(content_blocks[:5])])
outcome = "This should help you get started. Feel free to ask if you need further assistance."
return f"{confirmation}\n\n{steps}\n\n{outcome}"
def _structure_conversational_response(self, content_blocks: List[Dict[str, Any]]) -> str:
"""Structure a conversational response"""
if not content_blocks:
return "Thanks for chatting! How can I assist you today?"
# Combine content naturally
combined_content = " ".join([block['content'] for block in content_blocks])
return combined_content[:500] + "..." if len(combined_content) > 500 else combined_content
def _enhance_response_quality(self, response: str, intent: str) -> str:
"""Enhance response quality based on intent"""
# Add simple enhancements
enhanced = response
# Check for clarity
if len(response.split()) < 5:
enhanced += "\n\nWould you like me to expand on this?"
# Add intent-specific enhancements
if intent == "information_request" and "?" not in response:
enhanced += "\n\nIs there anything specific you'd like to know more about?"
return enhanced
def _extract_source_references(self, agent_outputs: List[Dict[str, Any]]) -> List[str]:
"""Extract source references from agent outputs"""
sources = []
for output in agent_outputs:
agent_id = output.get('agent_id', 'unknown')
sources.append(agent_id)
return list(set(sources)) # Remove duplicates
def _format_agent_outputs_for_synthesis(self, agent_outputs: List[Dict[str, Any]]) -> str:
"""Format agent outputs for LLM synthesis prompt"""
formatted = []
for i, output in enumerate(agent_outputs, 1):
agent_id = output.get('agent_id', 'unknown')
content = output.get('result', output.get('final_response', str(output)))
formatted.append(f"Agent {i} ({agent_id}): {content[:100]}...")
return "\n".join(formatted)
def _calculate_quality_metrics(self, synthesis_result: Dict[str, Any]) -> Dict[str, Any]:
"""Calculate quality metrics for synthesis"""
response = synthesis_result.get('final_response', '')
return {
"length": len(response),
"word_count": len(response.split()),
"coherence_score": synthesis_result.get('coherence_score', 0.7),
"source_count": len(synthesis_result.get('source_references', [])),
"has_structured_elements": bool(re.search(r'[•\d+\.]', response))
}
def _check_intent_alignment(self, synthesis_result: Dict[str, Any], intent_info: Dict[str, Any]) -> Dict[str, Any]:
"""Check if synthesis aligns with detected intent"""
alignment_score = 0.8 # Placeholder
return {
"intent_detected": intent_info.get('primary_intent'),
"alignment_score": alignment_score,
"alignment_verified": alignment_score > 0.7
}
def _identify_improvements(self, response: str) -> List[str]:
"""Identify opportunities to improve the response"""
improvements = []
if len(response) < 50:
improvements.append("Could be more detailed")
if "?" not in response and len(response.split()) < 100:
improvements.append("Consider adding examples")
return improvements
def _get_fallback_response(self, user_input: str, agent_outputs: List[Dict[str, Any]]) -> Dict[str, Any]:
"""Provide fallback response when synthesis fails"""
return {
"final_response": f"I apologize, but I'm having trouble generating a response. Your question was: {user_input[:100]}...",
"draft_response": "",
"source_references": [],
"coherence_score": 0.3,
"improvement_opportunities": ["System had synthesis error"],
"synthesis_method": "fallback",
"agent_id": self.agent_id,
"synthesis_quality_metrics": {"error": "synthesis_failed"},
"intent_alignment": {"error": "not_available"},
"error_handled": True
}
# Factory function for easy instantiation
def create_synthesis_agent(llm_router=None):
return ResponseSynthesisAgent(llm_router)
|