File size: 3,717 Bytes
f440f03 | 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 | """Datu priekšapstrāde."""
from __future__ import annotations
import json
import re
from typing import Any
def clean_text(text: str) -> str:
"""Notīra tekstu no nevēlamiem simboliem."""
text = re.sub(r"\s+", " ", text)
text = text.strip()
return text
def truncate(text: str, max_chars: int = 4096) -> str:
"""Apgriež tekstu līdz max_chars."""
return text[:max_chars]
def format_conversation(messages: list[dict[str, str]]) -> str:
"""Formatē sarunu kā vienu tekstu apmācībai."""
parts = []
for msg in messages:
role = msg.get("role", "user")
content = msg.get("content", "")
parts.append(f"<|{role}|>\n{content}\n<|end|>")
return "\n".join(parts)
def _preserve_block_text(value: Any) -> str:
return str(value or "").strip()
def _append_section(lines: list[str], title: str, value: Any) -> None:
if value is None:
return
if isinstance(value, str):
text = _preserve_block_text(value)
if text:
lines.append(f"{title}:\n{text}")
return
if isinstance(value, list):
items = [_preserve_block_text(item) for item in value if _preserve_block_text(item)]
if items:
lines.append(f"{title}:\n" + "\n".join(f"- {item}" for item in items))
return
if isinstance(value, dict) and value:
lines.append(f"{title}:\n{json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True)}")
def _format_structured_prompt_record(record: dict[str, Any]) -> str:
prompt = clean_text(str(record.get("prompt", "")))
user_sections = [prompt]
section_fields = (
("Repo konteksts", record.get("repo_context")),
("Mērķa fails", record.get("target_file")),
("Esošais vai kļūdainais kods", record.get("buggy_code")),
("Refactor vai diff konteksts", record.get("diff")),
("Papildu konteksts", record.get("context")),
("Pieņemšanas kritēriji", record.get("acceptance_criteria")),
("Testi", record.get("tests")),
("Robežgadījumi", record.get("edge_cases")),
)
for title, value in section_fields:
_append_section(user_sections, title, value)
metadata = record.get("metadata")
if metadata:
_append_section(user_sections, "Metadata", metadata)
messages = [
{"role": "user", "content": "\n\n".join(section for section in user_sections if section)}
]
completion = _preserve_block_text(record.get("completion"))
if completion:
messages.append({"role": "assistant", "content": completion})
elif metadata:
messages.append(
{
"role": "assistant",
"content": json.dumps(metadata, ensure_ascii=False, sort_keys=True),
}
)
return format_conversation(messages)
def record_to_training_text(record: dict[str, Any], max_chars: int = 4096) -> str:
"""Pārveido vienu HF dataset ierakstu uz tekstu kauzālai apmācībai."""
if "text" in record and isinstance(record["text"], str):
return truncate(clean_text(record["text"]), max_chars=max_chars)
if "user" in record or "assistant" in record:
messages = [
{"role": "user", "content": clean_text(str(record.get("user", "")))},
{"role": "assistant", "content": clean_text(str(record.get("assistant", "")))},
]
return truncate(format_conversation(messages), max_chars=max_chars)
if "prompt" in record:
return truncate(_format_structured_prompt_record(record), max_chars=max_chars)
serialized = json.dumps(record, ensure_ascii=False, sort_keys=True)
return truncate(clean_text(serialized), max_chars=max_chars)
|