What it does
Trained to take a single line (or short snippet) of buggy Python code and output a corrected
version, using the Muennighoff/python-bugs dataset
(wrong-operator bugs - e.g. / instead of +, < instead of ==).
Training details
- Base model:
Qwen/Qwen3-0.6B
- Method: LoRA (r=8, alpha=16, target modules:
q_proj, v_proj)
- Dataset:
Muennighoff/python-bugs, 500 training examples, 300 steps, batch size 4
- Final training loss: 1.5433 (started at 1.9057)
Because this is a short demo run rather than a full training job, treat the outputs below as
illustrative of the pipeline, not as evidence of strong code-fixing ability - a real production
version of this would use more steps/examples.
Example before/after (from this specific run)
Example 1
Buggy input:
def zunionstore(self, dest, keys, aggregate=None):
"""
Union multiple sorted sets specified by ``keys`` into
a new sorted set, ``dest``. Scores in the destination will be
aggregated based on the ``aggregate``, or SUM if none is provided.
"""
return self._zaggregate('ZUNIONSTORE', dest, self, aggregate)
Expected fix:
def zunionstore(self, dest, keys, aggregate=None):
"""
Union multiple sorted sets specified by ``keys`` into
a new sorted set, ``dest``. Scores in the destination will be
aggregated based on the ``aggregate``, or SUM if none is provided.
"""
return self._zaggregate('ZUNIONSTORE', dest, keys, aggregate)
Model output:
def zunionstore(self, dest, keys, aggregate=None):
"""
Union multiple sorted sets specified by ``keys`` into
a new sorted set, ``dest``. Scores in the destination will be
aggregated based on the ``aggregate``, or SUM if none is provided.
"""
return self._zaggregate('ZUNIONSTORE', dest, self, aggregate)
Example 2
Buggy input:
def wait(self, num_replicas, timeout):
"""
Redis synchronous replication
That returns the number of replicas that processed the query when
we finally have at least ``num_replicas``, or when the ``timeout`` was
reached.
"""
return self.execute_command('WAIT', self, timeout)
Expected fix:
def wait(self, num_replicas, timeout):
"""
Redis synchronous replication
That returns the number of replicas that processed the query when
we finally have at least ``num_replicas``, or when the ``timeout`` was
reached.
"""
return self.execute_command('WAIT', num_replicas, timeout)
Model output:
def wait(self, num_replicas, timeout):
"""
Redis synchronous replication
That returns the number of replicas that processed the query when
we finally have at least ``num_replicas``, or when the ``timeout`` was
reached.
"""
return self.execute_command('WAIT', self, timeout)
Example 3
Buggy input:
def tab_insert(self, e):
'''Insert a tab character. '''
cursor = min(self.l_buffer.point, len(self.l_buffer.line_buffer))
ws = ' ' * (ws.tabstop - (cursor % self.tabstop))
self.insert_text(ws)
Expected fix:
def tab_insert(self, e):
'''Insert a tab character. '''
cursor = min(self.l_buffer.point, len(self.l_buffer.line_buffer))
ws = ' ' * (self.tabstop - (cursor % self.tabstop))
self.insert_text(ws)
Model output:
def tab_insert(self, e):
'''Insert a tab character. '''
cursor = min(self.l_buffer.point, len(self.l_buffer.line_buffer))
ws = ' ' * (ws.tabstop - (cursor % self.tabstop))
self.insert_text(ws)
Example 4
Buggy input:
def update(self, headers):
"This method is a public interface for a throttle storage class"
self.prune()
if 'x-throttling-control' not in headers:
return
status = self.parse_throttle(headers['x-throttling-control'])
retry_after = int(headers.get('retry-after', 0))
sql, values = status.convert(status, retry_after)
with self.db:
self.db.execute(sql, values)
Expected fix:
def update(self, headers):
"This method is a public interface for a throttle storage class"
self.prune()
if 'x-throttling-control' not in headers:
return
status = self.parse_throttle(headers['x-throttling-control'])
retry_after = int(headers.get('retry-after', 0))
sql, values = self.convert(status, retry_after)
with self.db:
self.db.execute(sql, values)
Model output:
def update(self, headers):
"This method is a public interface for a throttle storage class"
self.prune()
if 'x-throttling-control' not in headers:
return
status = self.parse_throttle(headers['x-throttling-control'])
retry_after = int(headers.get('retry-after', 0))
sql, values = status.convert(status, retry_after)
with self.db:
self.db.execute(sql, values)
Usage
from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import PeftModel
base = AutoModelForCausalLM.from_pretrained("Qwen/Qwen3-0.6B")
model = PeftModel.from_pretrained(base, "GregMillard/greg-pllabs-demo-1785681463")
tokenizer = AutoTokenizer.from_pretrained("GregMillard/greg-pllabs-demo-1785681463")