File size: 7,664 Bytes
975da6d | 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 | """
Prepare training data from AWS DevOps Agent documentation markdown files.
Reads markdown files from training/data/raw/ and generates instruction-response
pairs in JSONL format for fine-tuning.
Generates multiple Q&A pair types:
1. Full-page summary pairs
2. Section-level Q&A pairs
3. Targeted question variants per section
"""
import json
import re
from pathlib import Path
RAW_DIR = Path("training/data/raw")
OUTPUT_FILE = Path("training/data/train.jsonl")
SYSTEM_MSG = """/no_think
You are Tech Advisor, an expert on AWS cloud services with deep knowledge of AWS DevOps Agent.
You have comprehensive knowledge of AWS DevOps Agent including:
- What it is and how it works (Agent Spaces, topology, dual-console architecture)
- Key features: autonomous incident response, proactive prevention, on-demand SRE tasks
- Integrations: CloudWatch, Datadog, Dynatrace, New Relic, Splunk, Grafana, PagerDuty, GitHub, GitLab, Azure DevOps, ServiceNow, Slack
- GA features: Azure/on-prem support, Triage Agent, Learned/Custom Skills, Code Indexing, Private Connections
- Pricing: $0.0083 per agent-second, free trial details, AWS Support credits
- Getting started: Agent Spaces, connecting tools, running investigations
- Security: encryption, customer managed keys, IdP integration, CloudTrail auditing
- Available regions: US East, US West, Frankfurt, Ireland, Sydney, Tokyo
Be concise and structured. Use bullet points where appropriate. Provide accurate, detailed answers."""
def extract_sections(markdown: str) -> list[dict]:
"""Split a markdown document into sections by headers."""
sections = []
current_title = "Overview"
current_content = []
current_level = 0
for line in markdown.split("\n"):
header_match = re.match(r'^(#{1,4})\s+(.+)', line)
if header_match:
if current_content:
content_text = "\n".join(current_content).strip()
if content_text:
sections.append({
"title": current_title,
"content": content_text,
"level": current_level,
})
current_level = len(header_match.group(1))
current_title = header_match.group(2).strip()
current_content = []
else:
current_content.append(line)
if current_content:
content_text = "\n".join(current_content).strip()
if content_text:
sections.append({
"title": current_title,
"content": content_text,
"level": current_level,
})
return [s for s in sections if len(s["content"]) > 30]
def clean_topic_name(filename: str) -> str:
"""Convert filename to a readable topic name."""
name = filename.replace(".md", "")
name = re.sub(r'^(about-aws-devops-agent-|aws-devops-agent-|getting-started-with-aws-devops-agent-|working-with-devops-agent-|configuring-capabilities-for-aws-devops-agent-|connecting-telemetry-sources-|connecting-to-ticketing-and-chat-|connecting-to-cicd-pipelines-|connecting-azure-|custom-agents-|interfacing-with-the-devops-agent-|integrating-devops-agent-into-event-driven-applications-using-amazon-eventbridge-)', '', name)
name = name.replace("-", " ").replace("_", " ")
return name
def generate_question_variants(title: str, topic: str) -> list[str]:
"""Generate natural question variants for a section."""
title_lower = title.lower()
questions = []
if any(w in title_lower for w in ["what is", "about", "overview"]):
questions.extend([
f"What is {topic} in AWS DevOps Agent?",
f"Explain {topic} in AWS DevOps Agent.",
f"Tell me about {topic}.",
])
elif any(w in title_lower for w in ["getting started", "creating", "setup", "setting up"]):
questions.extend([
f"How do I set up {topic} in AWS DevOps Agent?",
f"Walk me through {title.lower()} for AWS DevOps Agent.",
f"What are the steps to {title.lower()}?",
])
elif any(w in title_lower for w in ["connecting", "integrat"]):
questions.extend([
f"How do I connect {topic} to AWS DevOps Agent?",
f"What's the process for integrating {topic} with AWS DevOps Agent?",
f"How does AWS DevOps Agent work with {topic}?",
])
elif any(w in title_lower for w in ["security", "encryption", "iam", "authentication"]):
questions.extend([
f"How does {topic} work in AWS DevOps Agent?",
f"What security features does AWS DevOps Agent provide for {topic}?",
f"Tell me about {topic} for AWS DevOps Agent.",
])
elif any(w in title_lower for w in ["pricing", "cost", "quota"]):
questions.extend([
f"What are the {topic} for AWS DevOps Agent?",
f"How much does AWS DevOps Agent cost?",
f"What are the limits and {topic} for AWS DevOps Agent?",
])
else:
questions.extend([
f"What is {title} in AWS DevOps Agent?",
f"Tell me about {title} in AWS DevOps Agent.",
f"How does {title} work in AWS DevOps Agent?",
])
return questions
def generate_pairs_from_doc(filepath: Path) -> list[dict]:
"""Generate training pairs from a single documentation file."""
content = filepath.read_text().strip()
if not content or len(content) < 50:
return []
topic = clean_topic_name(filepath.name)
pairs = []
# Pair 1: Full document as a comprehensive answer
full_question = f"Give me a comprehensive overview of {topic} in AWS DevOps Agent."
if len(content) > 200:
pairs.append({
"messages": [
{"role": "system", "content": SYSTEM_MSG},
{"role": "user", "content": full_question},
{"role": "assistant", "content": content},
]
})
# Pair 2: Direct "what is" question with full content
pairs.append({
"messages": [
{"role": "system", "content": SYSTEM_MSG},
{"role": "user", "content": f"What is {topic}?"},
{"role": "assistant", "content": content},
]
})
# Section-level pairs with question variants
sections = extract_sections(content)
for section in sections:
if len(section["content"]) < 50:
continue
questions = generate_question_variants(section["title"], topic)
for q in questions[:2]:
pairs.append({
"messages": [
{"role": "system", "content": SYSTEM_MSG},
{"role": "user", "content": q},
{"role": "assistant", "content": section["content"]},
]
})
return pairs
def main():
if not RAW_DIR.exists():
print(f"Create {RAW_DIR} and add AWS documentation markdown files.")
RAW_DIR.mkdir(parents=True, exist_ok=True)
return
md_files = list(RAW_DIR.glob("*.md"))
if not md_files:
print(f"No markdown files found in {RAW_DIR}")
return
all_pairs = []
for filepath in sorted(md_files):
pairs = generate_pairs_from_doc(filepath)
all_pairs.extend(pairs)
print(f" {filepath.name}: {len(pairs)} training pairs")
OUTPUT_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(OUTPUT_FILE, "w") as f:
for pair in all_pairs:
f.write(json.dumps(pair) + "\n")
print(f"\nTotal: {len(all_pairs)} training pairs written to {OUTPUT_FILE}")
if __name__ == "__main__":
main()
|