412 lines
20 KiB
Python
412 lines
20 KiB
Python
"""
|
|
Placeholder Factory
|
|
Centralized placeholder extraction functions for all workflow modes.
|
|
Each function corresponds to a {{KEY:PLACEHOLDER_NAME}} in prompt templates.
|
|
|
|
NAMING CONVENTION:
|
|
- All functions follow pattern: extract{PlaceholderName}()
|
|
- Placeholder names are in UPPER_CASE with underscores
|
|
- Function names are in camelCase
|
|
|
|
MAPPING TABLE (keys → function) with usage [taskplan | actionplan | react]:
|
|
{{KEY:USER_PROMPT}} -> extractUserPrompt() [taskplan, actionplan, react]
|
|
{{KEY:USER_LANGUAGE}} -> extractUserLanguage() [actionplan, react]
|
|
{{KEY:WORKFLOW_HISTORY}} -> extractWorkflowHistory() [taskplan, actionplan, react]
|
|
{{KEY:AVAILABLE_CONNECTIONS_INDEX}} -> extractAvailableConnectionsIndex() [actionplan, react]
|
|
{{KEY:AVAILABLE_CONNECTIONS_SUMMARY}} -> extractAvailableConnectionsSummary() []
|
|
{{KEY:AVAILABLE_DOCUMENTS_SUMMARY}} -> extractAvailableDocumentsSummary() [taskplan, actionplan, react]
|
|
{{KEY:AVAILABLE_DOCUMENTS_INDEX}} -> extractAvailableDocumentsIndex() [react]
|
|
{{KEY:AVAILABLE_METHODS}} -> extractAvailableMethods() [actionplan, react]
|
|
{{KEY:REVIEW_CONTENT}} -> extractReviewContent() [actionplan, react]
|
|
{{KEY:PREVIOUS_ACTION_RESULTS}} -> extractPreviousActionResults() [react]
|
|
{{KEY:LEARNINGS_AND_IMPROVEMENTS}} -> extractLearningsAndImprovements() [react]
|
|
{{KEY:LATEST_REFINEMENT_FEEDBACK}} -> extractLatestRefinementFeedback() [react]
|
|
|
|
Following placeholders are populated directly by prompt builders with according context in promptGenerationActionsReact module:
|
|
- ACTION_OBJECTIVE,
|
|
- SELECTED_ACTION,
|
|
- ACTION_SIGNATURE
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from typing import Dict, Any, List
|
|
from modules.datamodels.datamodelChat import ChatDocument
|
|
|
|
logger = logging.getLogger(__name__)
|
|
from modules.workflows.processing.shared.methodDiscovery import (methods, discoverMethods)
|
|
|
|
def extractUserPrompt(context: Any) -> str:
|
|
"""Extract user prompt from context. Maps to {{KEY:USER_PROMPT}}.
|
|
Prefer the cleaned intent stored on the services object if available via context.
|
|
Fallback to the task_step objective.
|
|
"""
|
|
try:
|
|
# Prefer services.currentUserPrompt when accessible through context
|
|
services = getattr(context, 'services', None)
|
|
if services and getattr(services, 'currentUserPrompt', None):
|
|
return services.currentUserPrompt
|
|
except Exception:
|
|
pass
|
|
|
|
if hasattr(context, 'task_step') and context.task_step:
|
|
return context.task_step.objective or 'No request specified'
|
|
return 'No request specified'
|
|
|
|
def extractWorkflowHistory(service: Any, context: Any) -> str:
|
|
"""Extract workflow history from context. Maps to {{KEY:WORKFLOW_HISTORY}}
|
|
Reverse-chronological, enriched with message summaries and document labels.
|
|
"""
|
|
# Prefer explicit workflow on context; else fall back to services.workflow
|
|
workflow = None
|
|
try:
|
|
if hasattr(context, 'workflow') and context.workflow:
|
|
workflow = context.workflow
|
|
elif hasattr(service, 'workflow') and service.workflow:
|
|
workflow = service.workflow
|
|
except Exception:
|
|
workflow = None
|
|
|
|
if workflow:
|
|
history = getPreviousRoundContext(service, workflow)
|
|
return history or "No previous workflow rounds available"
|
|
return "No previous workflow rounds available"
|
|
|
|
def extractAvailableMethods(service: Any) -> str:
|
|
"""Extract available methods for action planning. Maps to {{KEY:AVAILABLE_METHODS}}"""
|
|
try:
|
|
# Get the methods dictionary directly from the global methods variable
|
|
if not methods:
|
|
discoverMethods(service)
|
|
|
|
# Create a flat JSON format with compound action names for better AI parsing
|
|
available_actions_json = {}
|
|
for methodName, methodInfo in methods.items():
|
|
# Convert MethodAi -> ai, MethodDocument -> document, etc.
|
|
shortName = methodName.replace('Method', '').lower()
|
|
|
|
for actionName, actionInfo in methodInfo['actions'].items():
|
|
# Create compound action name: method.action
|
|
compoundActionName = f"{shortName}.{actionName}"
|
|
# Get the action description
|
|
action_description = actionInfo.get('description', f"Execute {actionName} action")
|
|
available_actions_json[compoundActionName] = action_description
|
|
|
|
return json.dumps(available_actions_json, indent=2, ensure_ascii=False)
|
|
except Exception as e:
|
|
logger.error(f"Error extracting available methods: {str(e)}")
|
|
return json.dumps({}, indent=2, ensure_ascii=False)
|
|
|
|
def extractUserLanguage(service: Any) -> str:
|
|
"""Extract user language from service. Maps to {{KEY:USER_LANGUAGE}}"""
|
|
return service.user.language if service and service.user else 'en'
|
|
|
|
|
|
def _computeMessageSummary(msg) -> str:
|
|
"""Create a concise summary for a ChatMessage with documents only.
|
|
Fields: documentCount, roundNumber, documentsLabel, document names, message (full), success flag.
|
|
"""
|
|
try:
|
|
docs = getattr(msg, 'documents', []) or []
|
|
if not docs:
|
|
return "" # Only summarize messages that contain documents
|
|
document_count = len(docs)
|
|
round_number = getattr(msg, 'roundNumber', None) or 0
|
|
label = getattr(msg, 'documentsLabel', None) or ""
|
|
# Collect ALL document names (supports ChatDocument objects and dicts)
|
|
doc_names = []
|
|
for d in docs:
|
|
name = None
|
|
try:
|
|
if isinstance(d, dict):
|
|
# For dict objects, try multiple possible field names
|
|
name = d.get('fileName') or d.get('documentName') or d.get('name') or d.get('filename')
|
|
else:
|
|
# For ChatDocument objects, use fileName field
|
|
name = getattr(d, 'fileName', None) or getattr(d, 'documentName', None) or getattr(d, 'name', None) or getattr(d, 'filename', None)
|
|
except Exception:
|
|
name = None
|
|
doc_names.append(name or "(unnamed)")
|
|
# Format document names in brackets
|
|
if doc_names:
|
|
names_part = f"({', '.join(doc_names)})"
|
|
else:
|
|
names_part = "(no documents)"
|
|
|
|
# Don't truncate the message - show full content
|
|
user_message = (getattr(msg, 'message', '') or '').strip().replace("\n", " ")
|
|
# Read success from ChatMessage.success field
|
|
success_flag = getattr(msg, 'success', None)
|
|
success_text = "success=True" if success_flag is True else ("success=False" if success_flag is False else "success=Unknown")
|
|
label_part = f" label='{label}'" if label else ""
|
|
|
|
# Add learning/feedback if available
|
|
learning_part = ""
|
|
if hasattr(msg, 'summary') and msg.summary and 'learnings' in msg.summary.lower():
|
|
learning_part = " | learnings available"
|
|
|
|
return f"Round {round_number}: {document_count} docs {names_part}{label_part} | {success_text}{learning_part} | msg='{user_message}'"
|
|
except Exception:
|
|
return ""
|
|
|
|
def getMessageSummary(msg) -> str:
|
|
"""Return existing ChatMessage.summary or compute, set, and return it (documents only)."""
|
|
try:
|
|
if getattr(msg, 'summary', None):
|
|
return msg.summary
|
|
summary = _computeMessageSummary(msg)
|
|
# Persist in-memory only; caller can store if desired
|
|
if summary:
|
|
try:
|
|
setattr(msg, 'summary', summary)
|
|
except Exception:
|
|
pass
|
|
return summary
|
|
except Exception:
|
|
return ""
|
|
|
|
def getPreviousRoundContext(services, workflow: Any) -> str:
|
|
"""Get enriched context:
|
|
- Reverse-chronological ordering
|
|
- Current round first (newest → oldest), then older rounds
|
|
- Only messages with documents summarized
|
|
- Include available documents snapshot at end
|
|
"""
|
|
try:
|
|
if not workflow:
|
|
return "No previous round context available"
|
|
|
|
lines: List[str] = []
|
|
|
|
# Reverse-chronological, current round first
|
|
try:
|
|
msgs = getattr(workflow, 'messages', []) or []
|
|
current_round = getattr(workflow, 'currentRound', None)
|
|
current_round_msgs: List[Any] = []
|
|
previous_round_msgs: List[Any] = []
|
|
for m in msgs:
|
|
if current_round is not None and getattr(m, 'roundNumber', None) == current_round:
|
|
current_round_msgs.append(m)
|
|
else:
|
|
previous_round_msgs.append(m)
|
|
|
|
for m in reversed(current_round_msgs):
|
|
s = getMessageSummary(m)
|
|
if s:
|
|
lines.append(f"- {s}")
|
|
for m in reversed(previous_round_msgs):
|
|
s = getMessageSummary(m)
|
|
if s:
|
|
lines.append(f"- {s}")
|
|
except Exception:
|
|
pass
|
|
|
|
# Include available documents snapshot at end
|
|
try:
|
|
if hasattr(services, 'workflow'):
|
|
docs_index = services.workflow.getAvailableDocuments(workflow)
|
|
if docs_index and docs_index != "No documents available":
|
|
doc_count = docs_index.count("docItem:") # Only count actual documents, not document list labels
|
|
lines.append(f"Available documents: {doc_count}")
|
|
except Exception:
|
|
pass
|
|
|
|
if not lines:
|
|
return "No previous round context available"
|
|
return "\n".join(lines)
|
|
except Exception as e:
|
|
logger.error(f"Error getting previous round context: {str(e)}")
|
|
return "Error retrieving previous round context"
|
|
|
|
def extractReviewContent(context: Any) -> str:
|
|
"""Extract review content for result validation. Maps to {{KEY:REVIEW_CONTENT}}"""
|
|
try:
|
|
if hasattr(context, 'action_results') and context.action_results:
|
|
# Build result summary
|
|
result_summary = ""
|
|
for i, result in enumerate(context.action_results):
|
|
result_summary += f"\nRESULT {i+1}:\n"
|
|
result_summary += f" Success: {result.success}\n"
|
|
if result.error:
|
|
result_summary += f" Error: {result.error}\n"
|
|
|
|
if result.documents:
|
|
result_summary += f" Documents: {len(result.documents)} document(s)\n"
|
|
for doc in result.documents:
|
|
# Extract all available metadata without content
|
|
doc_metadata = {
|
|
"name": getattr(doc, 'fileName', None) or getattr(doc, 'documentName', 'Unknown'),
|
|
"mimeType": getattr(doc, 'mimeType', 'Unknown'),
|
|
"size": getattr(doc, 'size', 'Unknown'),
|
|
"created": getattr(doc, 'created', 'Unknown'),
|
|
"modified": getattr(doc, 'modified', 'Unknown'),
|
|
"typeGroup": getattr(doc, 'typeGroup', 'Unknown'),
|
|
"documentId": getattr(doc, 'documentId', 'Unknown'),
|
|
"reference": getattr(doc, 'reference', 'Unknown')
|
|
}
|
|
# Remove 'Unknown' values to keep it clean
|
|
doc_metadata = {k: v for k, v in doc_metadata.items() if v != 'Unknown'}
|
|
result_summary += f" - {json.dumps(doc_metadata, indent=6, ensure_ascii=False)}\n"
|
|
else:
|
|
result_summary += f" Documents: None\n"
|
|
|
|
return result_summary
|
|
elif hasattr(context, 'observation') and context.observation:
|
|
# For observation data, show full content but handle documents specially
|
|
if isinstance(context.observation, dict):
|
|
# Create a copy to modify
|
|
obs_copy = context.observation.copy()
|
|
|
|
# If there are previews with documents, show only metadata
|
|
if 'previews' in obs_copy and isinstance(obs_copy['previews'], list):
|
|
for preview in obs_copy['previews']:
|
|
if isinstance(preview, dict) and 'snippet' in preview:
|
|
# Replace snippet with metadata indicator
|
|
preview['snippet'] = f"[Content: {len(preview.get('snippet', ''))} characters]"
|
|
|
|
return json.dumps(obs_copy, indent=2, ensure_ascii=False)
|
|
else:
|
|
return json.dumps(context.observation, ensure_ascii=False)
|
|
elif hasattr(context, 'step_result') and context.step_result and 'observation' in context.step_result:
|
|
# For observation data in step_result, show full content but handle documents specially
|
|
observation = context.step_result['observation']
|
|
if isinstance(observation, dict):
|
|
# Create a copy to modify
|
|
obs_copy = observation.copy()
|
|
|
|
# If there are previews with documents, show only metadata
|
|
if 'previews' in obs_copy and isinstance(obs_copy['previews'], list):
|
|
for preview in obs_copy['previews']:
|
|
if isinstance(preview, dict) and 'snippet' in preview:
|
|
# Replace snippet with metadata indicator
|
|
preview['snippet'] = f"[Content: {len(preview.get('snippet', ''))} characters]"
|
|
|
|
return json.dumps(obs_copy, indent=2, ensure_ascii=False)
|
|
else:
|
|
return json.dumps(observation, ensure_ascii=False)
|
|
else:
|
|
return "No review content available"
|
|
except Exception as e:
|
|
logger.error(f"Error extracting review content: {str(e)}")
|
|
return "No review content available"
|
|
|
|
def extractPreviousActionResults(context: Any) -> str:
|
|
"""Extract previous action results for learning context. Maps to {{KEY:PREVIOUS_ACTION_RESULTS}}"""
|
|
try:
|
|
if not hasattr(context, 'previous_action_results') or not context.previous_action_results:
|
|
return "No previous actions executed yet"
|
|
|
|
results = []
|
|
for i, result in enumerate(context.previous_action_results[-5:], 1): # Last 5 results
|
|
if hasattr(result, 'resultLabel') and hasattr(result, 'status'):
|
|
status = "SUCCESS" if result.status == "completed" else "FAILED"
|
|
results.append(f"Action {i}: {result.resultLabel} - {status}")
|
|
if hasattr(result, 'error') and result.error:
|
|
results.append(f" Error: {result.error}")
|
|
|
|
return "\n".join(results) if results else "No previous actions executed yet"
|
|
except Exception as e:
|
|
logger.error(f"Error extracting previous action results: {str(e)}")
|
|
return "No previous actions executed yet"
|
|
|
|
def extractLearningsAndImprovements(context: Any) -> str:
|
|
"""Extract learnings and improvements from previous actions. Maps to {{KEY:LEARNINGS_AND_IMPROVEMENTS}}"""
|
|
try:
|
|
learnings = []
|
|
|
|
# Get improvements from context
|
|
if hasattr(context, 'improvements') and context.improvements and isinstance(context.improvements, list):
|
|
learnings.append("IMPROVEMENTS:")
|
|
for improvement in context.improvements[-3:]: # Last 3 improvements
|
|
learnings.append(f"- {improvement}")
|
|
|
|
# Get failure patterns
|
|
if hasattr(context, 'failure_patterns') and context.failure_patterns and isinstance(context.failure_patterns, list):
|
|
learnings.append("FAILURE PATTERNS TO AVOID:")
|
|
for pattern in context.failure_patterns[-3:]: # Last 3 patterns
|
|
learnings.append(f"- {pattern}")
|
|
|
|
# Get successful actions
|
|
if hasattr(context, 'successful_actions') and context.successful_actions and isinstance(context.successful_actions, list):
|
|
learnings.append("SUCCESSFUL APPROACHES:")
|
|
for action in context.successful_actions[-3:]: # Last 3 successful
|
|
learnings.append(f"- {action}")
|
|
|
|
return "\n".join(learnings) if learnings else "No learnings available yet"
|
|
except Exception as e:
|
|
logger.error(f"Error extracting learnings and improvements: {str(e)}")
|
|
return "No learnings available yet"
|
|
|
|
def extractLatestRefinementFeedback(context: Any) -> str:
|
|
"""Extract the latest refinement feedback. Maps to {{KEY:LATEST_REFINEMENT_FEEDBACK}}"""
|
|
try:
|
|
if not hasattr(context, 'previous_review_result') or not context.previous_review_result or not isinstance(context.previous_review_result, list):
|
|
return "No previous refinement feedback available"
|
|
|
|
# Get the most recent refinement decision
|
|
latest_decision = context.previous_review_result[-1]
|
|
if not isinstance(latest_decision, dict):
|
|
return "No previous refinement feedback available"
|
|
|
|
feedback_parts = []
|
|
|
|
# Add decision and reason
|
|
decision = latest_decision.get('decision', 'unknown')
|
|
reason = latest_decision.get('reason', 'No reason provided')
|
|
feedback_parts.append(f"Latest Decision: {decision}")
|
|
feedback_parts.append(f"Reason: {reason}")
|
|
|
|
# Add any specific feedback or suggestions
|
|
if 'feedback' in latest_decision:
|
|
feedback_parts.append(f"Feedback: {latest_decision['feedback']}")
|
|
|
|
if 'suggestions' in latest_decision:
|
|
feedback_parts.append(f"Suggestions: {latest_decision['suggestions']}")
|
|
|
|
return "\n".join(feedback_parts)
|
|
except Exception as e:
|
|
logger.error(f"Error extracting latest refinement feedback: {str(e)}")
|
|
return "No previous refinement feedback available"
|
|
|
|
def extractAvailableDocumentsSummary(service: Any, context: Any) -> str:
|
|
"""Summary of available documents (count only)."""
|
|
try:
|
|
documents = service.workflow.getAvailableDocuments(context.workflow)
|
|
if documents and documents != "No documents available":
|
|
# Count only actual documents, not list labels
|
|
doc_count = documents.count("docItem:")
|
|
return f"{doc_count} documents available from previous tasks"
|
|
return "No documents available"
|
|
except Exception as e:
|
|
logger.error(f"Error getting document summary: {str(e)}")
|
|
return "No documents available"
|
|
|
|
def extractAvailableDocumentsIndex(service: Any, context: Any) -> str:
|
|
"""Index of available documents with detailed references for parameter generation."""
|
|
try:
|
|
return service.workflow.getAvailableDocuments(context.workflow)
|
|
except Exception as e:
|
|
logger.error(f"Error getting document index: {str(e)}")
|
|
return "No documents available"
|
|
|
|
def extractAvailableConnectionsSummary(service: Any) -> str:
|
|
"""Summary of available connections (count only)."""
|
|
try:
|
|
connections = service.workflow.getConnectionReferenceList()
|
|
if connections:
|
|
return f"{len(connections)} connections available"
|
|
return "No connections available"
|
|
except Exception as e:
|
|
logger.error(f"Error getting connection summary: {str(e)}")
|
|
return "No connections available"
|
|
|
|
def extractAvailableConnectionsIndex(service: Any) -> str:
|
|
"""Index of available connections with detailed references for parameter generation."""
|
|
try:
|
|
connections = service.workflow.getConnectionReferenceList()
|
|
if connections:
|
|
return '\n'.join(f"- {conn}" for conn in connections)
|
|
return "No connections available"
|
|
except Exception as e:
|
|
logger.error(f"Error getting connection index: {str(e)}")
|
|
return "No connections available"
|