github/gpt-pilot
1
1
Fork 0
mirror of https://github.com/Pythagora-io/gpt-pilot.git synced 2026-01-09 21:27:53 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add feature

Browse Source
This commit is contained in:
mijauexe
2025-05-27 17:36:10 +02:00
parent 74f7c8c41d
commit 574423137b
13 changed files with 200 additions and 195 deletions
Download Patch File Download Diff File Expand all files Collapse all files

194
core/agents/spec_writer.py
View File

@@ -1,21 +1,18 @@
import secrets
from core.agents.base import BaseAgent
from core.agents.convo import AgentConvo
from core.agents.response import AgentResponse, ResponseType
from core.config import SPEC_WRITER_AGENT_NAME
from core.config import DESCRIBE_FILES_AGENT_NAME, SPEC_WRITER_AGENT_NAME
from core.config.actions import SPEC_CHANGE_FEATURE_STEP_NAME, SPEC_CHANGE_STEP_NAME, SPEC_CREATE_STEP_NAME
from core.db.models import Complexity
from core.db.models.project_state import IterationStatus
from core.llm.parser import StringParser
from core.log import get_logger
from core.telemetry import telemetry
from core.templates.registry import PROJECT_TEMPLATES
from core.ui.base import ProjectStage
# If the project description is less than this, perform an analysis using LLM
ANALYZE_THRESHOLD = 1500
# URL to the wiki page with tips on how to write a good project description
INITIAL_PROJECT_HOWTO_URL = (
"https://github.com/Pythagora-io/gpt-pilot/wiki/How-to-write-a-good-initial-project-description"
)
log = get_logger(__name__)
@@ -29,32 +26,88 @@ class SpecWriter(BaseAgent):
return await self.update_spec(iteration_mode=True)
elif self.prev_response and self.prev_response.type == ResponseType.UPDATE_SPECIFICATION:
return await self.update_spec(iteration_mode=False)
elif not self.current_state.knowledge_base.user_options.get("project_description", ""):
return await self.initialize_spec_and_project()
else:
return await self.initialize_spec()
return await self.change_spec()
async def initialize_spec(self) -> AgentResponse:
async def apply_template(self):
"""
Applies a template to the frontend.
"""
options = self.current_state.knowledge_base.user_options
if options["auth_type"] == "api_key" or options["auth_type"] == "none":
template_name = "vite_react_swagger"
else:
template_name = "vite_react"
template_class = PROJECT_TEMPLATES.get(template_name)
if not template_class:
log.error(f"Project template not found: {template_name}")
return
template = template_class(
options,
self.state_manager,
self.process_manager,
)
self.state_manager.template["template"] = template
log.info(f"Applying project template: {template.name}")
summary = await template.apply()
self.next_state.relevant_files = template.relevant_files
self.next_state.modified_files = {}
self.next_state.specification.template_summary = summary
async def initialize_spec_and_project(self) -> AgentResponse:
self.next_state.action = SPEC_CREATE_STEP_NAME
await self.ui.send_project_stage({"stage": ProjectStage.PROJECT_DESCRIPTION})
# if self.state_manager.user_options is not None and not self.state_manager.user_options.get("project_description", ""):
user_description = await self.ask_question(
"Please describe the app you want to build.",
allow_empty=False,
full_screen=True,
)
description = user_description.text.strip()
complexity = await self.check_prompt_complexity(description)
llm = self.get_llm(SPEC_WRITER_AGENT_NAME, stream_output=True)
convo = AgentConvo(self).template(
"build_full_specification",
initial_prompt=description,
auth=self.current_state.knowledge_base.user_options.get("auth", False),
)
await self.ui.start_important_stream()
llm_assisted_description = await llm(convo)
await self.ui.send_project_stage({"stage": ProjectStage.PROJECT_NAME})
llm = self.get_llm(DESCRIBE_FILES_AGENT_NAME)
convo = AgentConvo(self).template(
"project_name",
description=llm_assisted_description,
)
llm_response: str = await llm(convo, temperature=0)
project_name = llm_response.strip().replace(" ", "_").replace("-", "_")
await self.state_manager.rename_project(self.state_manager.project.id, project_name)
self.state_manager.file_system.root = (
self.state_manager.get_full_parent_project_root() + "/" + self.state_manager.project.folder_name
)
self.state_manager.file_system.ignore_matcher.root_path = self.state_manager.file_system.root
self.process_manager.root_dir = self.state_manager.file_system.root
self.next_state.knowledge_base.user_options["original_description"] = description
self.next_state.knowledge_base.user_options["project_description"] = llm_assisted_description
# await self.state_manager.commit()
return AgentResponse.done(self)
async def change_spec(self) -> AgentResponse:
llm = self.get_llm(SPEC_WRITER_AGENT_NAME, stream_output=True)
llm_assisted_description = self.current_state.knowledge_base.user_options["project_description"]
description = self.current_state.knowledge_base.user_options["original_description"]
while True:
user_done_with_description = await self.ask_question(
"Are you satisfied with the project description?",
@@ -74,18 +127,31 @@ class SpecWriter(BaseAgent):
allow_empty=False,
)
convo = convo.template("add_to_specification", user_message=user_add_to_spec.text.strip())
convo = (
AgentConvo(self).template(
"build_full_specification",
initial_prompt=llm_assisted_description.strip(),
)
).template("add_to_specification", user_message=user_add_to_spec.text.strip())
if len(convo.messages) > 6:
convo.slice(1, 4)
await self.ui.start_important_stream()
llm_assisted_description = await llm(convo)
convo = convo.template(
"build_full_specification",
auth=self.current_state.knowledge_base.user_options.get("auth", False),
initial_prompt=llm_assisted_description.strip(),
)
llm = self.get_llm(SPEC_WRITER_AGENT_NAME)
convo = AgentConvo(self).template(
"need_auth",
description=llm_assisted_description,
)
llm_response: str = await llm(convo, temperature=0)
auth = llm_response.strip().lower() == "yes"
if auth:
self.next_state.knowledge_base.user_options["auth"] = auth
self.next_state.knowledge_base.user_options["jwt_secret"] = secrets.token_hex(32)
self.next_state.knowledge_base.user_options["refresh_token_secret"] = secrets.token_hex(32)
# if we reload the project from the 1st project state, state_manager.template will be None
if self.state_manager.template:
@@ -97,6 +163,8 @@ class SpecWriter(BaseAgent):
self.next_state.specification = self.current_state.specification.clone()
self.next_state.specification.original_description = description
self.next_state.specification.description = llm_assisted_description
complexity = await self.check_prompt_complexity(description)
self.next_state.specification.complexity = complexity
telemetry.set("initial_prompt", description)
@@ -131,6 +199,10 @@ class SpecWriter(BaseAgent):
}
]
if not self.state_manager.async_tasks:
self.state_manager.async_tasks = []
await self.apply_template()
return AgentResponse.done(self)
async def update_spec(self, iteration_mode) -> AgentResponse:
@@ -185,91 +257,3 @@ class SpecWriter(BaseAgent):
llm_response: str = await llm(convo, temperature=0, parser=StringParser())
log.info(f"Complexity check response: {llm_response}")
return llm_response.lower()
async def analyze_spec(self, spec: str) -> str:
msg = (
"Your project description seems a bit short. "
"The better you can describe the project, the better Pythagora will understand what you'd like to build.\n\n"
f"Here are some tips on how to better describe the project: {INITIAL_PROJECT_HOWTO_URL}\n\n"
"Let's start by refining your project idea:"
)
await self.send_message(msg)
llm = self.get_llm(SPEC_WRITER_AGENT_NAME, stream_output=True)
convo = AgentConvo(self).template("ask_questions").user(spec)
n_questions = 0
n_answers = 0
while True:
response: str = await llm(convo)
if len(response) > 500:
# The response is too long for it to be a question, assume it's the updated spec
confirm = await self.ask_question(
("Would you like to change or add anything? Write it out here."),
allow_empty=True,
buttons={"continue": "No thanks, the spec looks good"},
)
if confirm.cancelled or confirm.button == "continue" or confirm.text == "":
updated_spec = response.strip()
await telemetry.trace_code_event(
"spec-writer-questions",
{
"num_questions": n_questions,
"num_answers": n_answers,
"new_spec": updated_spec,
},
)
return updated_spec
convo.user(confirm.text)
else:
convo.assistant(response)
n_questions += 1
user_response = await self.ask_question(
response,
buttons={"skip": "Skip this question", "skip_all": "No more questions"},
verbose=False,
)
if user_response.cancelled or user_response.button == "skip_all":
convo.user(
"This is enough clarification, you have all the information. "
"Please output the spec now, without additional comments or questions."
)
response: str = await llm(convo)
confirm = await self.ask_question(
("Would you like to change or add anything? Write it out here."),
allow_empty=True,
buttons={"continue": "No thanks, the spec looks good"},
)
if confirm.cancelled or confirm.button == "continue" or confirm.text == "":
updated_spec = response.strip()
await telemetry.trace_code_event(
"spec-writer-questions",
{
"num_questions": n_questions,
"num_answers": n_answers,
"new_spec": updated_spec,
},
)
return updated_spec
convo.user(confirm.text)
continue
n_answers += 1
if user_response.button == "skip":
convo.user("Skip this question.")
continue
else:
convo.user(user_response.text)
async def review_spec(self, desc: str, spec: str) -> str:
convo = AgentConvo(self).template("review_spec", desc=desc, spec=spec)
llm = self.get_llm(SPEC_WRITER_AGENT_NAME)
llm_response: str = await llm(convo, temperature=0)
additional_info = llm_response.strip()
if additional_info and len(additional_info) > 6:
spec += "\n\nAdditional info/examples:\n\n" + additional_info
await self.send_message(f"\n\nAdditional info/examples:\n\n {additional_info}")
return spec