| 1 | from typing import Any |
| 2 | |
| 3 | from helpers.extension import Extension, extensible |
| 4 | from helpers import projects |
| 5 | from agent import Agent, LoopData |
| 6 | |
| 7 | |
| 8 | class ProjectPrompt(Extension): |
| 9 | |
| 10 | async def execute( |
| 11 | self, |
| 12 | system_prompt: list[str] = [], |
| 13 | loop_data: LoopData = LoopData(), |
| 14 | **kwargs: Any, |
| 15 | ): |
| 16 | if not self.agent: |
| 17 | return |
| 18 | prompt = await build_prompt(self.agent, loop_data=loop_data) |
| 19 | if prompt: |
| 20 | system_prompt.append(prompt) |
| 21 | |
| 22 | |
| 23 | @extensible |
| 24 | async def build_prompt(agent: Agent, loop_data: LoopData | None = None) -> str: |
| 25 | result = agent.read_prompt("agent.system.projects.main.md") |
| 26 | project_name = agent.context.get_data(projects.CONTEXT_DATA_KEY_PROJECT) |
| 27 | if loop_data: |
| 28 | loop_data.protocol_persistent.pop("agents_md_instructions", None) |
| 29 | loop_data.protocol_persistent.pop("project_instructions", None) |
| 30 | if project_name: |
| 31 | project_vars = projects.build_system_prompt_vars(project_name) |
| 32 | if loop_data and project_vars.get("include_agents_md", True): |
| 33 | agents_md_protocol = projects.build_agents_md_protocol(project_name) |
| 34 | if agents_md_protocol: |
| 35 | loop_data.protocol_persistent["agents_md_instructions"] = ( |
| 36 | agents_md_protocol |
| 37 | ) |
| 38 | if loop_data and project_vars.get("project_instructions"): |
| 39 | loop_data.protocol_persistent["project_instructions"] = agent.read_prompt( |
| 40 | "agent.protocol.projects.instructions.md", |
| 41 | **project_vars, |
| 42 | ) |
| 43 | result += "\n\n" + agent.read_prompt( |
| 44 | "agent.system.projects.active.md", **project_vars |
| 45 | ) |
| 46 | else: |
| 47 | result += "\n\n" + agent.read_prompt("agent.system.projects.inactive.md") |
| 48 | return result |