File size: 3,062 Bytes
c85c557 | 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 | """
Synthetic Trajectory Generator for Gemma Fine-Tuning.
Generates multi-step tool execution sequences for agent training.
"""
import json
from pathlib import Path
def create_full_trajectory(symbol: str, file_path: str, bug_desc: str) -> dict:
return {
"messages": [
{
"role": "user",
"content": f"Investigate and resolve issue in symbol '{symbol}': {bug_desc}"
},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"type": "function",
"function": {
"name": "code_graph_search",
"parameters": {"symbol_name": symbol}
}
}
]
},
{
"role": "tool",
"content": json.dumps({
"status": "success",
"query": symbol,
"definitions": [{"file": file_path, "type": "class", "lineno": 10}]
})
},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"type": "function",
"function": {
"name": "read_scope",
"parameters": {"file_path": file_path, "start_line": 10, "end_line": 35}
}
}
]
},
{
"role": "tool",
"content": json.dumps({
"status": "success",
"file_path": file_path,
"content": "def execute(self):\n pass\n"
})
},
{
"role": "assistant",
"content": f"Located `{symbol}` in `{file_path}`. Verified scope and ready for resolution."
}
]
}
def main():
output_path = Path("data/train_trajectories.jsonl")
output_path.parent.mkdir(parents=True, exist_ok=True)
samples = [
create_full_trajectory("FileOperations", "src/tools/file_ops.py", "Boundary check validation failure."),
create_full_trajectory("RepositoryASTParser", "src/tools/ast_parser.py", "AST scope extraction missing line bounds."),
create_full_trajectory("CodeGraphIndexer", "src/tools/code_graph.py", "Symbol indexing cache invalidation issue."),
create_full_trajectory("PyTestExecutor", "src/tools/executor.py", "Subprocess execution timeout handling."),
create_full_trajectory("ContextCompactor", "src/agent/compactor.py", "Token limit calculation drift.")
]
with open(output_path, "w", encoding="utf-8") as f:
for sample in samples:
f.write(json.dumps(sample) + "\n")
print(f"Generated {len(samples)} multi-step training trajectories at {output_path}")
if __name__ == "__main__":
main()
|