adiuvAI/api
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor system prompt variables for clarity and consistency across agent setup and runner modules

Browse Source
This commit is contained in:
Roberto Musso
2026-04-07 00:23:41 +02:00
parent 1ce1d492b0
commit aa8bcbf0d8
3 changed files with 17 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
app/core/agent_runner.py
View File

@@ -102,7 +102,7 @@ _DOMAIN_DESCRIPTIONS: dict[str, str] = {
),
}
_STEP1_SYSTEM_PROMPT = """\
_BATCH_FILE_CLASSIFIER_PROMPT = """\
You are a file classifier for a freelance project management tool.
Your job is to match a file to an existing project and identify which data domains to extract.
@@ -133,7 +133,7 @@ Respond ONLY with a JSON object — no markdown, no explanation:
# ── Step 2: Processing prompt ─────────────────────────────────────────────
_PROCESSING_SYSTEM_PROMPT = """\
_BATCH_PROCESSING_PROMPT = """\
You are a data extraction assistant for a freelance project management tool.
Your task: extract structured data from the file content and persist it using the available tools.
@@ -162,7 +162,7 @@ Domains to extract: {data_types}
# ── Cloud processing prompt (kept separate for cloud agent) ───────────────
_CLOUD_PROCESSING_PROMPT = """\
_BATCH_CLOUD_PROCESSING_PROMPT = """\
You are a data extraction and management assistant for a freelance project
management tool.
@@ -560,7 +560,7 @@ async def _classify_file(
)
step1_template, step1_prompt_obj = get_prompt_or_fallback(
"batch_file_classifier", _STEP1_SYSTEM_PROMPT
"batch_file_classifier", _BATCH_FILE_CLASSIFIER_PROMPT
)
system = step1_template.format(
domain_definitions=domain_definitions,
@@ -774,7 +774,7 @@ async def run_local_agent(
existing_context = "\n\n".join(existing_blocks)
step2_template, step2_prompt_obj = get_prompt_or_fallback(
"batch_processing", _PROCESSING_SYSTEM_PROMPT
"batch_processing", _BATCH_PROCESSING_PROMPT
)
system_prompt = step2_template.format(
existing_context=existing_context,
@@ -995,7 +995,7 @@ async def run_cloud_agent(
items_processed += 1
cloud_template, cloud_prompt_obj = get_prompt_or_fallback(
"batch_cloud_processing", _CLOUD_PROCESSING_PROMPT
"batch_cloud_processing", _BATCH_CLOUD_PROCESSING_PROMPT
)
processing_prompt = cloud_template.format(
data_types=", ".join(config.data_types),

18
app/core/deep_agent.py
View File

@@ -28,7 +28,7 @@ logger = logging.getLogger(__name__)
FloatingDomainType = Literal["task", "timeline", "project", "node"]
FloatingDomainSection = Literal["task", "timeline", "note"]
_HOME_SINGLE_AGENT_SYSTEM = (
_HOME_SYSTEM_PROMPT = (
"You are the home assistant with direct access to all tools: tasks, projects, notes, timelines, and memory tools. "
"Always use tools for factual data retrieval before answering. "
"When the user asks to remember, forget, or update what you know about them, use memory tools. "
@@ -41,7 +41,7 @@ _HOME_SINGLE_AGENT_SYSTEM = (
"For upcoming tasks, after tag lines add a short recommendation based on due date and priority."
)
_FLOATING_SINGLE_AGENT_SYSTEM = (
_FLOATING_SYSTEM_PROMPT = (
"You are the floating assistant with direct access to all tools: tasks, projects, notes, timelines, and memory tools. "
"Stay focused on the floating scope in context.scope and answer concisely. "
"Return plain text only. Do not output XML/HTML-like tags such as <task>, <project>, <note>, <timeline>, or any bracketed id tag wrappers. "
@@ -50,7 +50,7 @@ _FLOATING_SINGLE_AGENT_SYSTEM = (
"If context.context.resolved_project_id exists, use it as project_id for scoped list calls. "
)
_FLOATING_DOMAIN_CLASSIFIER_SYSTEM = (
_FLOATING_DOMAIN_CLASSIFIER_PROMPT = (
"You are a strict domain classifier for websocket floating requests. "
"Return ONLY a JSON object with keys: type, id, section. "
"Allowed type values: task, timeline, project, node. "
@@ -539,7 +539,7 @@ async def _infer_floating_domain(message: str, context: dict[str, Any]) -> dict[
try:
llm = get_llm()
classifier_messages = [
SystemMessage(content=_FLOATING_DOMAIN_CLASSIFIER_SYSTEM),
SystemMessage(content=_FLOATING_DOMAIN_CLASSIFIER_PROMPT),
HumanMessage(
content=(
f"Message:\n{message}\n\n"
@@ -549,7 +549,7 @@ async def _infer_floating_domain(message: str, context: dict[str, Any]) -> dict[
]
lf = get_langfuse()
_, classifier_prompt_obj = get_prompt_or_fallback(
"floating_domain_classifier", _FLOATING_DOMAIN_CLASSIFIER_SYSTEM
"floating_domain_classifier", _FLOATING_DOMAIN_CLASSIFIER_PROMPT
)
if lf:
with lf.start_as_current_observation(
@@ -851,7 +851,7 @@ async def _run_single_agent_stream(
async def run_home(user_id: str, message: str, context: dict[str, Any]) -> str:
prepared_context = await _prepare_context(message, context)
system_prompt, langfuse_prompt = get_prompt_or_fallback(
"home_system", _HOME_SINGLE_AGENT_SYSTEM
"home_system", _HOME_SYSTEM_PROMPT
)
response = await _run_single_agent(
user_id=user_id,
@@ -868,7 +868,7 @@ async def run_floating(user_id: str, message: str, context: dict[str, Any]) -> t
prepared_context = await _prepare_context(message, context)
domain = await _infer_floating_domain(message, prepared_context)
system_prompt, langfuse_prompt = get_prompt_or_fallback(
"floating_system", _FLOATING_SINGLE_AGENT_SYSTEM
"floating_system", _FLOATING_SYSTEM_PROMPT
)
response = await _run_single_agent(
user_id=user_id,
@@ -891,7 +891,7 @@ async def run_home_stream(
) -> AsyncGenerator[tuple[str, Any], None]:
prepared_context = await _prepare_context(message, context)
system_prompt, langfuse_prompt = get_prompt_or_fallback(
"home_system", _HOME_SINGLE_AGENT_SYSTEM
"home_system", _HOME_SYSTEM_PROMPT
)
text_chunks: list[str] = []
async for event in _run_single_agent_stream(
@@ -923,7 +923,7 @@ async def run_floating_stream(
yield "floating_domain", domain
system_prompt, langfuse_prompt = get_prompt_or_fallback(
"floating_system", _FLOATING_SINGLE_AGENT_SYSTEM
"floating_system", _FLOATING_SYSTEM_PROMPT
)
sanitizer = _FloatingStreamSanitizer()
emitted_sanitized = False