eaada4bc57
- 新增 steward_intent_registry，IntentDescriptor 统一描述意图的识别关键词、动作步骤构建、字段白名单与副作用集合，替代分散的 if/else - 新增 steward_intent_bootstrap 注册 expense_application 等意图；新增 steward_query_executors 提供差旅标准查询的无副作用执行与城市/席别标签化输出 - action_contracts/action_executor/graph_planner/intent_agent/model_plan_builder/planner_extraction/fallback 适配注册表，识别与执行分发自动从注册表取数 - 新增 intent_registry/query_executors 测试，更新 intent_agent 测试
101 lines
3.7 KiB
Python
101 lines
3.7 KiB
Python
from app.services.steward_intent_agent import (
|
||
STEWARD_INTENT_FUNCTION_NAME,
|
||
StewardIntentAgent,
|
||
)
|
||
from app.schemas.steward import StewardPlanRequest
|
||
|
||
|
||
class _NoToolCallRuntimeChatService:
|
||
def __init__(self) -> None:
|
||
self.kwargs = {}
|
||
|
||
def complete_with_tool_call(self, messages, **kwargs):
|
||
self.kwargs = kwargs
|
||
|
||
class _Result:
|
||
tool_call = None
|
||
|
||
@staticmethod
|
||
def calls_as_dicts():
|
||
return []
|
||
|
||
return _Result()
|
||
|
||
|
||
def test_steward_intent_tool_schema_supports_pending_flow_confirmation() -> None:
|
||
schema = StewardIntentAgent._build_intent_tool_schema(
|
||
["expense_type", "time_range", "location", "reason", "transport_mode"]
|
||
)
|
||
|
||
function_schema = schema["function"]
|
||
assert function_schema["name"] == STEWARD_INTENT_FUNCTION_NAME
|
||
properties = function_schema["parameters"]["properties"]
|
||
task_schema = properties["tasks"]["items"]
|
||
pending_schema = properties["pending_flow_confirmation"]
|
||
candidate_schema = pending_schema["properties"]["candidate_flows"]["items"]
|
||
|
||
assert task_schema["properties"]["requested_action"]["enum"] == [
|
||
"preview",
|
||
"save_draft",
|
||
"submit",
|
||
]
|
||
assert "requested_action" in task_schema["required"]
|
||
assert "pending_flow_confirmation" in properties
|
||
assert pending_schema["properties"]["status"]["enum"] == ["none", "pending"]
|
||
assert candidate_schema["properties"]["flow_id"]["enum"] == [
|
||
"travel_application",
|
||
"travel_reimbursement",
|
||
]
|
||
assert candidate_schema["properties"]["missing_fields"]["items"]["enum"] == [
|
||
"expense_type",
|
||
"time_range",
|
||
"location",
|
||
"reason",
|
||
"transport_mode",
|
||
]
|
||
|
||
|
||
def test_steward_intent_agent_uses_ten_second_timeout_and_three_attempts() -> None:
|
||
runtime_chat = _NoToolCallRuntimeChatService()
|
||
agent = StewardIntentAgent(runtime_chat)
|
||
|
||
agent.detect(
|
||
StewardPlanRequest(message="2026-02-20 至 2026-02-23，上海出差，火车，保存草稿。"),
|
||
base_date=__import__("datetime").date(2026, 6, 24),
|
||
canonical_fields=["expense_type", "time_range", "location", "reason", "transport_mode"],
|
||
)
|
||
|
||
assert runtime_chat.kwargs["timeout_seconds"] == 10
|
||
assert runtime_chat.kwargs["max_attempts"] == 3
|
||
assert runtime_chat.kwargs["use_failure_cooldown"] is False
|
||
|
||
|
||
def test_steward_intent_tool_schema_includes_query_task_type_from_registry() -> None:
|
||
"""function call schema 的 task_type enum 应从注册表动态生成,包含查询意图。"""
|
||
from app.services import steward_intent_bootstrap # noqa: F401 触发意图注册
|
||
|
||
schema = StewardIntentAgent._build_intent_tool_schema(
|
||
["expense_type", "time_range", "location", "reason", "transport_mode"]
|
||
)
|
||
task_schema = schema["function"]["parameters"]["properties"]["tasks"]["items"]
|
||
task_type_enum = task_schema["properties"]["task_type"]["enum"]
|
||
|
||
assert "expense_application" in task_type_enum
|
||
assert "reimbursement" in task_type_enum
|
||
assert "query_travel_standard" in task_type_enum
|
||
|
||
|
||
def test_steward_intent_system_prompt_mentions_query_intent_guidance() -> None:
|
||
"""system prompt 应包含查询意图的识别指引,避免被误识别为申请。"""
|
||
from app.services import steward_intent_bootstrap # noqa: F401 触发意图注册
|
||
|
||
messages = StewardIntentAgent._build_messages(
|
||
StewardPlanRequest(message="武汉出差标准是多少"),
|
||
base_date=__import__("datetime").date(2026, 6, 24),
|
||
canonical_fields=["location", "employee_grade"],
|
||
)
|
||
system_prompt = messages[0]["content"]
|
||
assert "query_travel_standard" in system_prompt
|
||
assert "差旅" in system_prompt
|
||
assert "住宿标准" in system_prompt
|