File size: 7,102 Bytes
ae1e90a
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
#!/usr/bin/env python3
"""
Task Contracts: Output Discipline for Execution Engine
Defines strict contracts for each task type to constrain LLM behavior
"""

from dataclasses import dataclass
from typing import Optional, Dict
from enum import Enum

class TaskType(Enum):
    """Supported task types with strict contracts"""
    GRAMMAR_FIX = "grammar_fix"
    MATH_CALC = "math_calc"
    CODE_GEN = "code_gen"
    EXPLANATION = "explanation"
    QUESTION_ANSWER = "qa"
    GENERAL = "general"

@dataclass
class TaskContract:
    """
    Execution contract for a task type
    
    This enforces:
    - Token budgets (prevent rambling)
    - Output style (prevent hallucination)
    - Creativity level (prevent story generation)
    - Format requirements (structured output)
    """
    task_type: TaskType
    max_tokens: int
    temperature: float  # 0.0 = deterministic, 1.0 = creative
    style: str  # "rewrite_only", "factual", "code_only", "concise"
    allow_explanation: bool
    output_format: str  # "single_line", "code_block", "paragraph"
    system_instruction: str
    
    def get_prompt_template(self, user_input: str, tool_result: Optional[str] = None) -> str:
        """Generate disciplined prompt based on contract"""
        
        if self.task_type == TaskType.GRAMMAR_FIX:
            return f"""Task: Grammar correction.
Rules:
- Output ONE corrected sentence only
- Do not add explanation
- Do not add commentary
- Preserve original meaning

Input: "{user_input}"
Corrected:"""

        elif self.task_type == TaskType.MATH_CALC:
            if tool_result:
                # Tool already gave answer - just format it
                return f"""The answer is: {tool_result}"""
            else:
                return f"""Task: Solve this math problem.
Rules:
- Show final answer only
- No explanation unless asked
- Use exact arithmetic

Problem: {user_input}
Answer:"""

        elif self.task_type == TaskType.CODE_GEN:
            return f"""Task: Write code.
Rules:
- Output code only
- Include brief comments
- No explanation outside code
- Follow best practices

Request: {user_input}
Code:
```python"""

        elif self.task_type == TaskType.EXPLANATION:
            return f"""Task: Explain concisely.
Rules:
- Maximum 3 sentences
- Focus on key points
- No unnecessary details

Question: {user_input}
Answer:"""

        elif self.task_type == TaskType.QUESTION_ANSWER:
            return f"""Task: Answer the question directly.
Rules:
- Direct answer first
- Brief explanation if needed
- No rambling

Question: {user_input}
Answer:"""

        else:  # GENERAL
            return user_input


# Task contract definitions
TASK_CONTRACTS = {
    TaskType.GRAMMAR_FIX: TaskContract(
        task_type=TaskType.GRAMMAR_FIX,
        max_tokens=48,
        temperature=0.0,
        style="rewrite_only",
        allow_explanation=False,
        output_format="single_line",
        system_instruction="You are a grammar correction tool. Output only the corrected sentence."
    ),
    
    TaskType.MATH_CALC: TaskContract(
        task_type=TaskType.MATH_CALC,
        max_tokens=32,
        temperature=0.0,
        style="factual",
        allow_explanation=False,
        output_format="single_line",
        system_instruction="You are a math solver. Output only the final answer."
    ),
    
    TaskType.CODE_GEN: TaskContract(
        task_type=TaskType.CODE_GEN,
        max_tokens=200,
        temperature=0.2,
        style="code_only",
        allow_explanation=False,
        output_format="code_block",
        system_instruction="You are a code generator. Output only clean, working code."
    ),
    
    TaskType.EXPLANATION: TaskContract(
        task_type=TaskType.EXPLANATION,
        max_tokens=128,
        temperature=0.3,
        style="concise",
        allow_explanation=True,
        output_format="paragraph",
        system_instruction="You are a concise explainer. Maximum 3 sentences."
    ),
    
    TaskType.QUESTION_ANSWER: TaskContract(
        task_type=TaskType.QUESTION_ANSWER,
        max_tokens=96,
        temperature=0.3,
        style="direct",
        allow_explanation=True,
        output_format="paragraph",
        system_instruction="You are a direct answerer. Answer first, explain briefly if needed."
    ),
    
    TaskType.GENERAL: TaskContract(
        task_type=TaskType.GENERAL,
        max_tokens=256,
        temperature=0.7,
        style="conversational",
        allow_explanation=True,
        output_format="paragraph",
        system_instruction="You are a helpful assistant."
    ),
}


def detect_task_type(query: str) -> TaskType:
    """Detect task type from query"""
    query_lower = query.lower()
    
    # Grammar fix
    if any(kw in query_lower for kw in ['fix this', 'correct this', 'grammar']):
        return TaskType.GRAMMAR_FIX
    
    # Math calculation
    if any(kw in query_lower for kw in ['calculate', 'compute', 'solve']) or \
       any(op in query for op in ['+', '-', '*', '/', '=']):
        return TaskType.MATH_CALC
    
    # Code generation
    if any(kw in query_lower for kw in ['write a', 'function', 'code', 'program', 'script']):
        return TaskType.CODE_GEN
    
    # Explanation
    if any(kw in query_lower for kw in ['explain', 'what is', 'how does']):
        return TaskType.EXPLANATION
    
    # Question answering
    if query.strip().endswith('?'):
        return TaskType.QUESTION_ANSWER
    
    return TaskType.GENERAL


def get_contract(task_type: TaskType) -> TaskContract:
    """Get contract for task type"""
    return TASK_CONTRACTS[task_type]


def should_skip_llm(task_type: TaskType, tool_result: Optional[str]) -> bool:
    """
    Determine if we should skip LLM and return tool result directly
    
    Hard-stop rule: If tool gives final answer, don't re-explain
    """
    if not tool_result:
        return False
    
    # Math: calculator gave answer, return it
    if task_type == TaskType.MATH_CALC:
        return True
    
    # Future: Add more hard-stop rules
    
    return False


def test_contracts():
    """Test task contracts"""
    print("="*70)
    print("🧪 Task Contract System Test")
    print("="*70)
    
    test_queries = [
        "Fix this: Me and him goes to store",
        "Calculate 123 * 456",
        "Write a Python function to reverse a string",
        "Explain what Python is",
        "What is 2 + 2?",
    ]
    
    for query in test_queries:
        task_type = detect_task_type(query)
        contract = get_contract(task_type)
        
        print(f"\n📝 Query: {query}")
        print(f"   Task Type: {task_type.value}")
        print(f"   Max Tokens: {contract.max_tokens}")
        print(f"   Temperature: {contract.temperature}")
        print(f"   Style: {contract.style}")
        print(f"   Format: {contract.output_format}")
        
        # Show prompt
        prompt = contract.get_prompt_template(query)
        print(f"\n   Prompt:")
        print("   " + "\n   ".join(prompt.split("\n")[:5]))
        print("   ...")


if __name__ == "__main__":
    test_contracts()