Python
```python
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional, List, Dict, Any
from enum import Enum
import re
class ReadinessState(str, Enum):
NOT_READY = "not_ready"
NEEDS_CLARIFICATION = "needs_clarification"
AWAITING_CONFIRMATION = "awaiting_confirmation"
READY = "ready"
@dataclass
class AmbiguousAnnotation:
text: str
possible_intents: List[str]
resolved: bool = False
resolution: Optional[str] = None
@dataclass
class CreativeIntentSchema:
"""Structured representation of user's creative intent."""
asset_type: str
mood: Optional[str] = None
scene_elements: List[Dict] = field(default_factory=list)
display_texts: List[Dict] = field(default_factory=list)
ambiguous_annotations: List[AmbiguousAnnotation] = field(default_factory=list)
turn_count: int = 0
user_confirmed_vision: bool = False
last_coach_summary: Optional[str] = None
def get_readiness(self) -> ReadinessState:
if self.turn_count == 0:
return ReadinessState.NOT_READY
unresolved = [a for a in self.ambiguous_annotations if not a.resolved]
if unresolved:
return ReadinessState.NEEDS_CLARIFICATION
if not self.user_confirmed_vision:
return ReadinessState.AWAITING_CONFIRMATION
return ReadinessState.READY
def is_ready(self) -> bool:
return self.get_readiness() == ReadinessState.READY
def get_clarification_questions(self) -> List[str]:
return [
f'Should "{a.text}" be rendered as an image or displayed as text?'
for a in self.ambiguous_annotations if not a.resolved
]
class IntentParser:
"""Parses messages to extract and update intent."""
CONFIRMATION_PATTERNS = [
r"\b(yes|yeah|sure|ok|perfect|great|looks good|exactly)\b",
r"\b(let's go|do it|generate|create it)\b",
]
def parse_initial_request(
self,
description: str,
asset_type: str,
mood: Optional[str] = None,
) -> CreativeIntentSchema:
schema = CreativeIntentSchema(asset_type=asset_type, mood=mood)
# Extract quoted text as display text
quoted = re.findall(r'"([^"]+)"', description)
for text in quoted:
schema.display_texts.append({"text": text})
if description and not quoted:
schema.scene_elements.append({"description": description})
return schema
def parse_user_message(
self,
message: str,
schema: CreativeIntentSchema,
) -> tuple[CreativeIntentSchema, bool]:
schema.turn_count += 1
is_confirmation = self._is_confirmation(message)
if is_confirmation:
schema.user_confirmed_vision = True
# Resolve ambiguities from user response
message_lower = message.lower()
for amb in schema.ambiguous_annotations:
if not amb.resolved:
if "text" in message_lower or "display" in message_lower:
amb.resolved = True
amb.resolution = "display_text"
elif "render" in message_lower or "image" in message_lower:
amb.resolved = True
amb.resolution = "render"
return schema, is_confirmation
def _is_confirmation(self, message: str) -> bool:
message_lower = message.lower().strip()
return any(re.search(p, message_lower) for p in self.CONFIRMATION_PATTERNS)
@dataclass
class StreamChunk:
type: str # "token", "intent_ready", "done", "error"
content: str = ""
metadata: Optional[Dict[str, Any]] = None
class CoachService:
"""Orchestrates coaching conversations."""
MAX_TURNS = 10
def __init__(self, llm_client, session_store):
self.llm = llm_client
self.sessions = session_store
self.parser = IntentParser()
async def start_session(
self,
user_id: str,
asset_type: str,
description: str,
mood: Optional[str] = None,
):
# Initialize intent schema
schema = self.parser.parse_initial_request(description, asset_type, mood)
# Build system prompt
system_prompt = f"""You are a creative coach helping users design {asset_type} assets.
RULES:
- Ask clarifying questions to understand their vision
- Summarize what you understand after each exchange
- When vision is clear, say [INTENT_READY]
- Never say [INTENT_READY] on first turn
- Focus on WHAT they want, not HOW"""
first_message = f'User wants to create a {asset_type}. Description: "{description}"'
# Stream LLM response
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": first_message},
]
full_response = ""
async for token in self.llm.stream_chat(messages):
full_response += token
yield StreamChunk(type="token", content=token)
# First turn is NEVER ready
yield StreamChunk(
type="intent_ready",
metadata={
"is_ready": False,
"readiness_state": ReadinessState.NOT_READY.value,
"clarification_questions": schema.get_clarification_questions(),
},
)
yield StreamChunk(type="done", metadata={"turns_remaining": self.MAX_TURNS - 1})
async def continue_chat(
self,
session_id: str,
message: str,
schema: CreativeIntentSchema,
):
if schema.turn_count >= self.MAX_TURNS:
yield StreamChunk(type="error", content="Turn limit reached")
return
schema, is_confirmation = self.parser.parse_user_message(message, schema)
# Stream response...
full_response = ""
async for token in self.llm.stream_chat([...]):
full_response += token
yield StreamChunk(type="token", content=token)
readiness = schema.get_readiness()
yield StreamChunk(
type="intent_ready",
metadata={
"is_ready": schema.is_ready(),
"readiness_state": readiness.value,
"is_confirmation": is_confirmation,
},
)
```
TypeScript
```typescript
enum ReadinessState {
NOT_READY = 'not_ready',
NEEDS_CLARIFICATION = 'needs_clarification',
AWAITING_CONFIRMATION = 'awaiting_confirmation',
READY = 'ready',
}
interface AmbiguousAnnotation {
text: string;
possibleIntents: string[];
resolved: boolean;
resolution?: string;
}
interface CreativeIntentSchema {
assetType: string;
mood?: string;
sceneElements: Array<{ description: string }>;
displayTexts: Array<{ text: string }>;
ambiguousAnnotations: AmbiguousAnnotation[];
turnCount: number;
userConfirmedVision: boolean;
}
function getReadiness(schema: CreativeIntentSchema): ReadinessState {
if (schema.turnCount === 0) return ReadinessState.NOT_READY;
const unresolved = schema.ambiguousAnnotations.filter(a => !a.resolved);
if (unresolved.length > 0) return ReadinessState.NEEDS_CLARIFICATION;
if (!schema.userConfirmedVision) return ReadinessState.AWAITING_CONFIRMATION;
return ReadinessState.READY;
}
const CONFIRMATION_PATTERNS = [
/\b(yes|yeah|sure|ok|perfect|great|looks good)\b/i,
/\b(let's go|do it|generate|create it)\b/i,
];
function isConfirmation(message: string): boolean {
return CONFIRMATION_PATTERNS.some(p => p.test(message));
}
```
|