Private
Public Access
0
0
Files
manual_slop/src/thinking_parser.py
T
ed 8f11340b38 refactor(consumers): migrate 85 'from src.models import' sites to direct subsystem imports
Per post_module_taxonomy_de_cruft_20260627 Phase 2 (FR7). Each
'from src.models import X' for a moved class is rewritten to
'from src.<destination> import X':

  Ticket, Track, WorkerContext, TrackState, TrackMetadata,
    ThinkingSegment, EMPTY_TRACK_STATE            -> src.mma
  ProjectContext, ProjectMeta, ProjectOutput, ProjectFiles,
    ProjectScreenshots, ProjectDiscussion, EMPTY_PROJECT_CONTEXT -> src.project
  FileItem, Preset, ContextPreset, ContextFileEntry,
    NamedViewPreset                                -> src.project_files
  Tool, ToolPreset                                 -> src.tool_presets
  BiasProfile                                      -> src.tool_bias
  TextEditorConfig, ExternalEditorConfig,
    EMPTY_TEXT_EDITOR_CONFIG                       -> src.external_editor
  Persona                                          -> src.personas
  WorkspaceProfile                                -> src.workspace_manager
  MCPServerConfig, MCPConfiguration, VectorStoreConfig,
    RAGConfig, load_mcp_config                      -> src.mcp_client

NOT touched (kept on src.models; Phase 3 or Phase 4 will move them):
  GenerateRequest, ConfirmRequest, DEFAULT_TOOL_CATEGORIES, Metadata, PROVIDERS

Migration was performed by the one-time script
scripts/tier2/artifacts/post_module_taxonomy_de_cruft_20260627/migrate_imports.py
which uses a class-to-module map and re.sub() to rewrite each
'from src.models import X' line.

Total: 85 import lines rewritten across 71 files.

Note: this commit depends on the v2 SHIPPED work
(origin/tier2/module_taxonomy_refactor_20260627) being merged into
this branch NEXT. On master (without the v2 SHIPPED commits), the
destination modules do not exist and these imports would fail.
2026-06-26 13:34:03 -04:00

55 lines
2.3 KiB
Python

import re
from typing import List, Tuple
from src.mma import ThinkingSegment
def parse_thinking_trace(text: str) -> Tuple[List[ThinkingSegment], str]:
"""
Parses thinking segments from text and returns (segments, response_content).
Support extraction of thinking traces from <thinking>...</thinking>, <thought>...</thought>,
<think>...</think> (half-width form), and blocks prefixed with Thinking:.
[C: tests/test_thinking_trace.py:test_parse_empty_response, tests/test_thinking_trace.py:test_parse_multiple_markers, tests/test_thinking_trace.py:test_parse_no_thinking, tests/test_thinking_trace.py:test_parse_text_thinking_prefix, tests/test_thinking_trace.py:test_parse_thinking_with_empty_response, tests/test_thinking_trace.py:test_parse_xml_thinking_tag, tests/test_thinking_trace.py:test_parse_xml_thought_tag, tests/test_thinking_trace.py:test_parse_half_width_think_tag]
"""
segments = []
# 1. Extract <thinking> and <thought> tags
current_text = text
# Combined pattern for tags
tag_pattern = re.compile(r'<(thinking|thought|think)>(.*?)</\1>', re.DOTALL | re.IGNORECASE)
def extract_tags(txt: str) -> Tuple[List[ThinkingSegment], str]:
found_segments = []
def replace_func(match):
marker = match.group(1).lower()
content = match.group(2).strip()
found_segments.append(ThinkingSegment(content=content, marker=marker))
return ""
remaining = tag_pattern.sub(replace_func, txt)
return found_segments, remaining
tag_segments, remaining = extract_tags(current_text)
segments.extend(tag_segments)
# 2. Extract Thinking: prefix
# This usually appears at the start of a block and ends with a double newline or a response marker.
thinking_colon_pattern = re.compile(r'(?:^|\n)Thinking:\s*(.*?)(?:\n\n|\nResponse:|\nAnswer:|$)', re.DOTALL | re.IGNORECASE)
def extract_colon_blocks(txt: str) -> Tuple[List[ThinkingSegment], str]:
found_segments = []
def replace_func(match):
content = match.group(1).strip()
if content: found_segments.append(ThinkingSegment(content=content, marker="Thinking:"))
return "\n\n"
res = thinking_colon_pattern.sub(replace_func, txt)
return found_segments, res
colon_segments, final_remaining = extract_colon_blocks(remaining)
segments.extend(colon_segments)
return segments, final_remaining.strip()