115 lines
4.5 KiB
Python
115 lines
4.5 KiB
Python
import logging
|
|
import json
|
|
import time
|
|
from typing import Dict, Any, List, Optional
|
|
from .models import ModelFactory
|
|
from .gitea import GiteaClient
|
|
from .task import Task
|
|
|
|
class AgentRunner:
|
|
"""
|
|
Agent runner for Wolf.
|
|
"""
|
|
def __init__(self, gitea_client: GiteaClient, config: Dict[str, Any]):
|
|
self.gitea = gitea_client
|
|
self.config = config
|
|
|
|
def execute_task(self, task: Task):
|
|
"""
|
|
Execute a task through a specified model.
|
|
"""
|
|
logging.info(f"Executing task {task.id} with model {task.assigned_model} via {task.assigned_provider}")
|
|
|
|
# 1. Craft the prompt
|
|
prompt = self._craft_prompt(task)
|
|
|
|
# 2. Get the model client
|
|
provider_config = self.config.get('providers', {}).get(task.assigned_provider, {})
|
|
client = ModelFactory.get_client(
|
|
task.assigned_provider,
|
|
api_key=provider_config.get('api_key'),
|
|
base_url=provider_config.get('base_url')
|
|
)
|
|
|
|
# 3. Generate response
|
|
try:
|
|
response = client.generate(prompt, task.assigned_model, system_prompt="You are an expert software engineer.")
|
|
logging.info(f"Generated response for task {task.id}")
|
|
except Exception as e:
|
|
logging.error(f"Error generating response for task {task.id}: {e}")
|
|
return None
|
|
|
|
# 4. Parse the response (assuming model returns a JSON or specific format)
|
|
# For simplicity, we'll assume the model returns a JSON with 'files' and 'commit_message'
|
|
# If not, we'll try to extract it.
|
|
try:
|
|
# Try to find JSON in the response
|
|
if "```json" in response:
|
|
json_str = response.split("```json")[1].split("```")[0].strip()
|
|
result = json.loads(json_str)
|
|
else:
|
|
result = json.loads(response)
|
|
except Exception as e:
|
|
logging.warning(f"Failed to parse JSON from response for task {task.id}: {e}")
|
|
# Fallback: Treat entire response as a single file change if possible
|
|
result = {
|
|
"files": [{"path": "solution.py", "content": response}],
|
|
"commit_message": f"Solution for task {task.id}"
|
|
}
|
|
|
|
# 5. Commit the output back to a feature branch
|
|
branch_name = f"wolf-task-{task.id}-{int(time.time())}"
|
|
try:
|
|
self.gitea.create_branch(task.repo_owner, task.repo_name, branch_name)
|
|
for file_change in result.get('files', []):
|
|
path = file_change['path']
|
|
content = file_change['content']
|
|
|
|
# Check if file exists to update or create
|
|
existing_file = self.gitea.get_file(task.repo_owner, task.repo_name, path)
|
|
if existing_file:
|
|
self.gitea.update_file(
|
|
task.repo_owner, task.repo_name, path, content,
|
|
branch_name, existing_file['sha'],
|
|
message=result.get('commit_message', f"Update {path} by Wolf")
|
|
)
|
|
else:
|
|
self.gitea.create_file(
|
|
task.repo_owner, task.repo_name, path, content,
|
|
branch_name, message=result.get('commit_message', f"Create {path} by Wolf")
|
|
)
|
|
|
|
# 6. Create PR
|
|
pr_title = f"Wolf Task: {task.title}"
|
|
pr_body = f"This PR was generated by Wolf using model {task.assigned_model} via {task.assigned_provider}.\n\nTask Description:\n{task.description}"
|
|
pr = self.gitea.create_pull_request(task.repo_owner, task.repo_name, pr_title, pr_body, branch_name)
|
|
logging.info(f"Created PR {pr['number']} for task {task.id}")
|
|
return pr
|
|
except Exception as e:
|
|
logging.error(f"Error committing changes or creating PR for task {task.id}: {e}")
|
|
return None
|
|
|
|
def _craft_prompt(self, task: Task):
|
|
"""
|
|
Craft the prompt for the model.
|
|
"""
|
|
return f"""
|
|
You are an expert software engineer. Your task is to solve the following coding problem:
|
|
|
|
Title: {task.title}
|
|
Description: {task.description}
|
|
|
|
Please provide the solution as a JSON object with the following structure:
|
|
{{
|
|
"files": [
|
|
{{
|
|
"path": "path/to/file.py",
|
|
"content": "file content here"
|
|
}}
|
|
],
|
|
"commit_message": "A meaningful commit message"
|
|
}}
|
|
|
|
Ensure the code is correct, well-documented, and follows best practices.
|
|
"""
|