refactor(backend): 更新 agentscope 提示词系统
This commit is contained in:
zl-q
@@ -1,9 +1,4 @@
|
||||
from core.agentscope.prompts.agent_prompt import (
|
||||
ROUTER_AGENT_INSTRUCTION,
|
||||
WORKER_AGENT_INSTRUCTION,
|
||||
build_agent_prompt,
|
||||
build_intent_user_prompt,
|
||||
)
|
||||
from core.agentscope.prompts.agent_prompt import build_agent_prompt
|
||||
from core.agentscope.prompts.system_prompt import build_system_prompt
|
||||
from core.agentscope.prompts.tool_prompt import build_tools_prompt
|
||||
|
||||
@@ -11,7 +6,4 @@ __all__ = [
|
||||
"build_agent_prompt",
|
||||
"build_system_prompt",
|
||||
"build_tools_prompt",
|
||||
"ROUTER_AGENT_INSTRUCTION",
|
||||
"WORKER_AGENT_INSTRUCTION",
|
||||
"build_intent_user_prompt",
|
||||
]
|
||||
|
||||
@@ -1,10 +1,8 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import Any
|
||||
from collections.abc import Callable
|
||||
|
||||
from schemas.agent.runtime_models import ResultType, RouterAgentOutput, TaskType
|
||||
from schemas.agent.system_agent import AgentType
|
||||
from schemas.agent.system_agent import AgentType, SystemAgentLLMConfig
|
||||
|
||||
|
||||
def _wrap_section(section: str, content: str) -> str:
|
||||
@@ -16,117 +14,96 @@ def _wrap_section(section: str, content: str) -> str:
|
||||
return f"{start}\n{body}\n{end}" if body else f"{start}\n{end}"
|
||||
|
||||
|
||||
def _enum_values(enum_cls: Any) -> str:
|
||||
return ", ".join(item.value for item in enum_cls)
|
||||
PromptRuleBuilder = Callable[[SystemAgentLLMConfig | None], list[str]]
|
||||
|
||||
|
||||
def _schema_json(model: type[Any]) -> str:
|
||||
return json.dumps(
|
||||
model.model_json_schema(), ensure_ascii=True, separators=(",", ":")
|
||||
)
|
||||
class AgentPromptRegistry:
|
||||
def __init__(self) -> None:
|
||||
self._builders: dict[AgentType, PromptRuleBuilder] = {}
|
||||
|
||||
def register(self, *, agent_type: AgentType, builder: PromptRuleBuilder) -> None:
|
||||
self._builders[agent_type] = builder
|
||||
|
||||
def build_rules(
|
||||
self,
|
||||
*,
|
||||
agent_type: AgentType,
|
||||
llm_config: SystemAgentLLMConfig | None,
|
||||
) -> list[str]:
|
||||
builder = self._builders.get(agent_type)
|
||||
if builder is None:
|
||||
builder = self._builders[AgentType.WORKER]
|
||||
return builder(llm_config)
|
||||
|
||||
|
||||
ROUTER_AGENT_INSTRUCTION = """
|
||||
[Router Agent]
|
||||
- Read the latest user input and produce a routing contract for downstream execution.
|
||||
- Return exactly one RouterAgentOutput JSON object.
|
||||
""".strip()
|
||||
|
||||
|
||||
WORKER_AGENT_INSTRUCTION = """
|
||||
[Worker Agent]
|
||||
- Execute or answer against the routed objective and available evidence.
|
||||
- Return exactly one worker output JSON object matching the runtime-injected schema.
|
||||
""".strip()
|
||||
|
||||
|
||||
def build_intent_user_prompt(
|
||||
*, user_input: str | list[dict[str, Any]]
|
||||
) -> str | list[dict[str, Any]]:
|
||||
instruction = "\n\n".join(
|
||||
[
|
||||
ROUTER_AGENT_INSTRUCTION,
|
||||
"[Output Schema]",
|
||||
_schema_json(RouterAgentOutput),
|
||||
]
|
||||
)
|
||||
if isinstance(user_input, list):
|
||||
return [{"type": "text", "text": instruction}, *user_input]
|
||||
return "\n\n".join([instruction, "[User Input]", user_input])
|
||||
|
||||
|
||||
def _router_role_rules() -> list[str]:
|
||||
def _config_rules(llm_config: SystemAgentLLMConfig | None) -> list[str]:
|
||||
if llm_config is None:
|
||||
return []
|
||||
context_mode = llm_config.context_messages.mode.value
|
||||
context_count = llm_config.context_messages.count
|
||||
tool_groups = [group.value for group in llm_config.enabled_tool_groups]
|
||||
return [
|
||||
"Router only: extract intent and route strategy; never answer user directly.",
|
||||
"Preserve intent in normalized_task_input.user_text; keep wording concise and faithful.",
|
||||
"Fill multimodal_summary only when image/attachment changes execution decisions.",
|
||||
"Return key_entities and constraints that are execution-relevant; low confidence -> omit rather than guess.",
|
||||
"Set execution_mode by complexity: onestep / tool_assisted / multistep.",
|
||||
"Set result_typing.primary to the most suitable response shape; use clarification_request only when required info is missing.",
|
||||
"Set ui.ui_mode and ui.ui_decision_reason based on whether structured UI improves actionability.",
|
||||
f"task_typing.primary must use one TaskType enum: {_enum_values(TaskType)}.",
|
||||
f"task_typing.secondary max 3 enums: {_enum_values(TaskType)}.",
|
||||
f"result_typing.primary must use one ResultType enum: {_enum_values(ResultType)}.",
|
||||
f"result_typing.secondary max 3 enums: {_enum_values(ResultType)}.",
|
||||
"[Runtime Config]",
|
||||
f"- context_messages.mode={context_mode}",
|
||||
f"- context_messages.count={context_count}",
|
||||
f"- enabled_tool_groups={','.join(tool_groups) if tool_groups else 'default'}",
|
||||
]
|
||||
|
||||
|
||||
def _worker_role_rules() -> list[str]:
|
||||
def _worker_rules(llm_config: SystemAgentLLMConfig | None) -> list[str]:
|
||||
return [
|
||||
"Worker only: execute routed objective without changing router intent.",
|
||||
"Treat router output as objective/constraints contract, not as a fully-materialized tool-args payload.",
|
||||
"Infer deterministic required tool arguments from contract fields, tool schema, and runtime context.",
|
||||
"Ask minimal clarification only when required arguments cannot be inferred safely.",
|
||||
"Ground every claim in available evidence and tool results; never fabricate execution state.",
|
||||
"Keep status/result_type/answer/key_points/suggested_actions/error internally consistent.",
|
||||
"[Worker Agent]",
|
||||
"- Process user context directly, identify intent, then execute or answer with available evidence.",
|
||||
"- Return exactly one agent output JSON object matching the runtime-injected schema.",
|
||||
"[Responsibilities]",
|
||||
"- Handle user request directly from conversation context.",
|
||||
"- Decide intent first, then choose direct answer, tool call, clarification, or refusal.",
|
||||
"- Prefer a direct answer when no tool result is required.",
|
||||
"- Call tools only when tool results are necessary to produce a correct answer.",
|
||||
"- Infer deterministic required tool arguments from context, tool schema, and runtime signals.",
|
||||
"- Ask minimal clarification only when required arguments cannot be inferred safely.",
|
||||
"- If request is unsafe or disallowed, return safe refusal with actionable alternative.",
|
||||
"- Ground every claim in evidence and tool outputs; never fabricate execution state.",
|
||||
"[Schema Guidance]",
|
||||
"- The output schema is injected at runtime; follow it exactly.",
|
||||
"- Do not add fields that are not present in the injected schema.",
|
||||
*_config_rules(llm_config),
|
||||
]
|
||||
|
||||
|
||||
def build_worker_contract_prompt(*, router_output: RouterAgentOutput) -> str:
|
||||
contract_json = json.dumps(
|
||||
router_output.model_dump(mode="json", exclude_none=True),
|
||||
ensure_ascii=False,
|
||||
separators=(",", ":"),
|
||||
)
|
||||
return "\n".join(
|
||||
[
|
||||
"[Worker Contract]",
|
||||
"- Keep routed objective unchanged.",
|
||||
"- Use normalized_task_input as objective text.",
|
||||
"- Use multimodal_summary/key_entities/constraints as execution evidence.",
|
||||
"- Infer deterministic missing required tool args from evidence + tool schema.",
|
||||
"- Ask clarification only when safe inference is impossible.",
|
||||
"[RouterAgentOutput]",
|
||||
contract_json,
|
||||
]
|
||||
)
|
||||
def _memory_rules(llm_config: SystemAgentLLMConfig | None) -> list[str]:
|
||||
return [
|
||||
"[Memory Agent]",
|
||||
"- Analyze conversation context and output structured memory-safe conclusions.",
|
||||
"- Return exactly one agent output JSON object matching the runtime-injected schema.",
|
||||
"[Responsibilities]",
|
||||
"- Focus on extracting durable user facts and preferences from context.",
|
||||
"- Keep outputs concise, deterministic, and evidence-backed.",
|
||||
"- Do not invent facts or hidden user intent.",
|
||||
"- Use tool calls only when required by explicit workflow and allowed tool groups.",
|
||||
"[Schema Guidance]",
|
||||
"- The output schema is injected at runtime; follow it exactly.",
|
||||
"- Do not add fields that are not present in the injected schema.",
|
||||
*_config_rules(llm_config),
|
||||
]
|
||||
|
||||
|
||||
def build_agent_prompt(*, agent_type: AgentType) -> str:
|
||||
AGENT_PROMPT_REGISTRY = AgentPromptRegistry()
|
||||
AGENT_PROMPT_REGISTRY.register(agent_type=AgentType.WORKER, builder=_worker_rules)
|
||||
AGENT_PROMPT_REGISTRY.register(agent_type=AgentType.MEMORY, builder=_memory_rules)
|
||||
|
||||
|
||||
def build_agent_prompt(
|
||||
*,
|
||||
agent_type: AgentType,
|
||||
llm_config: SystemAgentLLMConfig | None = None,
|
||||
) -> str:
|
||||
lines = [
|
||||
"[Agent Identity]",
|
||||
f"- type: {agent_type.value}",
|
||||
*AGENT_PROMPT_REGISTRY.build_rules(
|
||||
agent_type=agent_type,
|
||||
llm_config=llm_config,
|
||||
),
|
||||
]
|
||||
|
||||
if agent_type == AgentType.ROUTER:
|
||||
lines.extend([ROUTER_AGENT_INSTRUCTION, "[Responsibilities]"])
|
||||
lines.extend(f"- {rule}" for rule in _router_role_rules())
|
||||
lines.extend(
|
||||
[
|
||||
"[Schema Guidance]",
|
||||
"- Output target is RouterAgentOutput.",
|
||||
"- Keep routing output conservative when confidence is low; ask for clarification instead of guessing hidden facts.",
|
||||
]
|
||||
)
|
||||
else:
|
||||
lines.extend([WORKER_AGENT_INSTRUCTION, "[Responsibilities]"])
|
||||
lines.extend(f"- {rule}" for rule in _worker_role_rules())
|
||||
lines.extend(
|
||||
[
|
||||
"[Schema Guidance]",
|
||||
"- The worker output schema is injected at runtime; follow it exactly.",
|
||||
"- Do not add fields that are not present in the injected schema.",
|
||||
]
|
||||
)
|
||||
|
||||
return _wrap_section("agent", "\n".join(lines))
|
||||
|
||||
@@ -11,7 +11,7 @@ from core.agentscope.prompts.agent_prompt import (
|
||||
)
|
||||
from core.agentscope.prompts.route_prompt import build_frontend_route_prompt
|
||||
from core.agentscope.prompts.tool_prompt import build_tools_prompt
|
||||
from schemas.agent.system_agent import AgentType
|
||||
from schemas.agent.system_agent import AgentType, SystemAgentLLMConfig
|
||||
from schemas.agent.forwarded_props import ClientTimeContext
|
||||
from schemas.user.context import UserContext
|
||||
|
||||
@@ -202,12 +202,14 @@ def _build_route_section() -> str:
|
||||
def build_system_prompt(
|
||||
*,
|
||||
agent_type: AgentType,
|
||||
llm_config: SystemAgentLLMConfig | None,
|
||||
user_context: UserContext,
|
||||
now_utc: datetime,
|
||||
runtime_client_time: ClientTimeContext | None = None,
|
||||
extra_context: str | None = None,
|
||||
tools: Sequence[Tool | dict[str, Any]] | None = None,
|
||||
) -> str:
|
||||
include_route_section = agent_type == AgentType.WORKER
|
||||
sections: list[str | None] = [
|
||||
_build_identity_section(),
|
||||
_build_env_section(
|
||||
@@ -216,10 +218,11 @@ def build_system_prompt(
|
||||
runtime_client_time=runtime_client_time,
|
||||
extra_context=extra_context,
|
||||
),
|
||||
_build_route_section(),
|
||||
_build_route_section() if include_route_section else None,
|
||||
_build_safety_section(),
|
||||
build_agent_prompt(
|
||||
agent_type=agent_type,
|
||||
llm_config=llm_config,
|
||||
),
|
||||
build_tools_prompt(tools=tools) if tools else None,
|
||||
_build_output_rules(),
|
||||
|
||||
Reference in New Issue
Block a user