harmony

app.py

@@ -1,5 +1,4 @@
 from transformers import pipeline, TextIteratorStreamer
-import torch
 from threading import Thread
 import gradio as gr
 import spaces
@@ -10,8 +9,29 @@ from openai_harmony import (
     Role,
     Message,
     Conversation,
+    SystemContent,
+    DeveloperContent,
+    ReasoningEffort,
 )
 
+# regex config
+RE_REASONING = re.compile(r'(?i)Reasoning:\s*(low|medium|high)')
+RE_FINAL_MARKER = re.compile(r'(?i)assistantfinal')
+RE_ANALYSIS_PREFIX = re.compile(r'(?i)^analysis\s*')
+
+# I think for system prompt reasoning level OpenAI mentioned you should do parsing so here's
+def parse_reasoning_and_instructions(system_prompt: str):
+    instructions = system_prompt or "You are a helpful assistant."
+    match = RE_REASONING.search(instructions)
+    effort_key = match.group(1).lower() if match else 'medium'
+    effort = {
+        'low': ReasoningEffort.LOW,
+        'medium': ReasoningEffort.MEDIUM,
+        'high': ReasoningEffort.HIGH,
+    }.get(effort_key, ReasoningEffort.MEDIUM)
+    cleaned_instructions = RE_REASONING.sub('', instructions).strip()
+    return effort, cleaned_instructions
+
 model_id = "openai/gpt-oss-20b"
 
 pipe = pipeline(
@@ -31,42 +51,23 @@ def format_conversation_history(chat_history):
             content = content[0]["text"] if content and "text" in content[0] else str(content)
         messages.append({"role": role, "content": content})
     return messages
-#OpenAI's harmony format
-def build_harmony_conversation_from_messages(messages):
-    harmony_messages = []
-    for m in messages:
-        role = m["role"].lower()
-        content = m["content"]
-        if role == "system":
-            harmony_messages.append(
-                Message.from_role_and_content(
-                    Role.SYSTEM,
-                    content,
-                )
-            )
-        elif role == "user":
-            harmony_messages.append(
-                Message.from_role_and_content(
-                    Role.USER,
-                    content,
-                )
-            )
-        elif role == "assistant":
-            harmony_messages.append(
-                Message.from_role_and_content(
-                    Role.ASSISTANT,
-                    content,
-                )
-            )
-    return Conversation.from_messages(harmony_messages)
-    
+
 @spaces.GPU()
 def generate_response(input_data, chat_history, max_new_tokens, system_prompt, temperature, top_p, top_k, repetition_penalty):
     new_message = {"role": "user", "content": input_data}
-    system_message = [{"role": "system", "content": system_prompt}] if system_prompt else []
     processed_history = format_conversation_history(chat_history)
-    messages = system_message + processed_history + [new_message]
-    conversation = build_harmony_conversation_from_messages(messages)
+    effort, instructions = parse_reasoning_and_instructions(system_prompt)
+    system_content = SystemContent.new().with_reasoning_effort(effort)
+    developer_content = DeveloperContent.new().with_instructions(instructions)
+    harmony_messages = [
+        Message.from_role_and_content(Role.SYSTEM, system_content),
+        Message.from_role_and_content(Role.DEVELOPER, developer_content),
+    ]
+    
+    for m in processed_history + [new_message]:
+        role = Role.USER if m["role"] == "user" else Role.ASSISTANT
+        harmony_messages.append(Message.from_role_and_content(role, m["content"]))
+    conversation = Conversation.from_messages(harmony_messages)
     prompt_tokens = enc.render_conversation_for_completion(conversation, Role.ASSISTANT)
     prompt_text = pipe.tokenizer.decode(prompt_tokens, skip_special_tokens=False)
 
@@ -85,21 +86,20 @@ def generate_response(input_data, chat_history, max_new_tokens, system_prompt, t
     thread = Thread(target=pipe, args=(prompt_text,), kwargs=generation_kwargs)
     thread.start()
 
+    # parsing thinking
     thinking = ""
     final = ""
     started_final = False
     for chunk in streamer:
         if not started_final:
-            if "assistantfinal" in chunk.lower():
-                split_parts = re.split(r'(?i)assistantfinal', chunk, maxsplit=1)
-                thinking += split_parts[0]
-                final += split_parts[1]
+            parts = RE_FINAL_MARKER.split(chunk, maxsplit=1)
+            thinking += parts[0]
+            if len(parts) > 1:
+                final += parts[-1]
                 started_final = True
-            else:
-                thinking += chunk
         else:
             final += chunk
-        clean_thinking = re.sub(r'^analysis\s*', '', thinking, flags=re.I).strip()
+        clean_thinking = RE_ANALYSIS_PREFIX.sub('', thinking).strip()
         clean_final = final.strip()
         formatted = f"<details open><summary>Click to view Thinking Process</summary>\n\n{clean_thinking}\n\n</details>\n\n{clean_final}"
         yield formatted