diff --git a/app/model/goal_conversation.py b/app/model/goal_conversation.py new file mode 100644 index 0000000..9a1241f --- /dev/null +++ b/app/model/goal_conversation.py @@ -0,0 +1,26 @@ +"""Internal persistence model for conversational investment-goal extraction.""" + +from datetime import datetime +from typing import Any + +from sqlalchemy import JSON, BigInteger, DateTime, String +from sqlalchemy.orm import Mapped, mapped_column + +from app.model.base import Base + + +class AdvisorGoalConversationExtraction(Base): + """Append-only extraction snapshot; never exposed in the customer API.""" + + __tablename__ = "advisor_goal_conversation_extraction" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + message_id: Mapped[int] = mapped_column(BigInteger, unique=True, nullable=False) + session_id: Mapped[str] = mapped_column(String(64), nullable=False) + customer_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + extraction_version: Mapped[str] = mapped_column(String(32), nullable=False) + extracted_fields: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) + missing_fields: Mapped[list[str]] = mapped_column(JSON, nullable=False) + status: Mapped[str] = mapped_column(String(32), nullable=False) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) diff --git a/app/service/agent/implementations/advisor.py b/app/service/agent/implementations/advisor.py index 0be7e06..8a20c7a 100644 --- a/app/service/agent/implementations/advisor.py +++ b/app/service/agent/implementations/advisor.py @@ -4,6 +4,7 @@ from typing import Any from app.core.contracts import AgentDefinition, AgentRequest, CoreResult, RequestContext from app.service.agent.implementations.fund_query_demo import FundQueryDemoAgent +from app.service.goal_conversation_service import GoalConversationService class AdvisorAgent(FundQueryDemoAgent): @@ -27,6 +28,7 @@ class AdvisorAgent(FundQueryDemoAgent): "portfolio_analysis", "asset_allocation", "product_recommend", + "comparison", ), ) @@ -35,6 +37,12 @@ class AdvisorAgent(FundQueryDemoAgent): self._classified_intent is not None and self._classified_intent.intent == "investment_goal" ): + if GoalConversationService.should_collect(request.message): + goal_result = await GoalConversationService().process( + session_id=request.session_id, customer_id=int(context.user_id), + trace_id=context.trace_id, message=request.message, + ) + return CoreResult(text=GoalConversationService.customer_prompt(goal_result)) output = await self.call_tool( "query_investment_goal", {}, intent="investment_goal", context=context ) @@ -65,6 +73,11 @@ class AdvisorAgent(FundQueryDemoAgent): "recommend_products", {"limit": 3}, intent="product_recommend", context=context ) return CoreResult(text=self._describe_recommendation(output)) + if ( + self._classified_intent is not None + and self._classified_intent.intent == "comparison" + ): + return CoreResult(text="对比分析需要明确两个或多个场内基金产品，请提供基金代码或名称。") return await super().handle(request, context) @staticmethod diff --git a/app/service/agent_persistence_service.py b/app/service/agent_persistence_service.py index 3fdbbea..ee4b914 100644 --- a/app/service/agent_persistence_service.py +++ b/app/service/agent_persistence_service.py @@ -10,6 +10,7 @@ from app.core.errors import RunLeaseLostError from app.model.audit import InteractionAudit from app.model.conversation import ConversationMessage from app.model.platform import AgentRun, DomainEventOutbox, RequestIdempotency +from app.model.session import ConversationSession class AgentPersistenceService: @@ -58,6 +59,13 @@ class AgentPersistenceService: run.completed_at = now run.updated_at = now run.locked_until = None + session_row = await self.session.scalar(select(ConversationSession).where( + ConversationSession.session_id == run.session_id, + ConversationSession.user_id == run.user_id, + ).with_for_update()) + if session_row is not None and result.result.intent is not None: + session_row.last_intent = result.result.intent.intent + session_row.updated_at = now self.session.add(InteractionAudit( actor_type="agent", actor_id=run.user_id, target_customer_id=run.user_id, session_id=run.session_id, portal="agent", action_type="agent.run_completed", diff --git a/app/service/goal_conversation_service.py b/app/service/goal_conversation_service.py new file mode 100644 index 0000000..ce7a644 --- /dev/null +++ b/app/service/goal_conversation_service.py @@ -0,0 +1,228 @@ +"""Conversation-level investment-goal entity extraction and gap handling.""" + +import re +from datetime import UTC, datetime +from decimal import Decimal +from typing import Any, Protocol + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.infrastructure.db import SessionFactory +from app.model.audit import InteractionAudit +from app.model.conversation import ConversationMessage +from app.model.goal_conversation import AdvisorGoalConversationExtraction +from app.model.session import ConversationSession + +EXTRACTION_VERSION = "investment-goal-v1" +MAX_CLARIFICATION_ROUNDS = 10 +REQUIRED_FIELDS = ( + "annualized_return_lower_pct", + "annualized_return_upper_pct", + "max_drawdown_pct", + "liquidity_requirement", + "investment_horizon_months", + "benchmark_name", +) + +FIELD_LABELS = { + "annualized_return_lower_pct": "年化收益目标下限", + "annualized_return_upper_pct": "年化收益目标上限", + "max_drawdown_pct": "最大回撤容忍度", + "liquidity_requirement": "资金流动性要求", + "investment_horizon_months": "投资期限", + "benchmark_name": "业绩比较基准", +} + +_NUMBER = r"\d+(?:\.\d+)?" + + +class SessionFactoryLike(Protocol): + def __call__(self) -> AsyncSession: ... + + +def _number(value: str) -> float: + return float(Decimal(value)) + + +def extract_goal_entities(text: str) -> dict[str, Any]: + """Extract explicit goal facts from one user message. + + This parser intentionally accepts only explicit financial statements. It does not + infer a risk level, expected return, or benchmark from vague language. + """ + result: dict[str, Any] = {} + return_match = re.search( + rf"(?:年化)?(?:收益目标|收益率|收益)\s*(?:为|在|约为|目标为)?\s*({_NUMBER})\s*%?\s*" + rf"(?:至|到|[-~～])\s*({_NUMBER})\s*%?", + text, + re.IGNORECASE, + ) + if return_match: + lower, upper = map(_number, return_match.groups()) + result["annualized_return_lower_pct"] = min(lower, upper) + result["annualized_return_upper_pct"] = max(lower, upper) + else: + single_return = re.search( + rf"(?:年化)?(?:收益目标|收益率|收益)\s*" + rf"(?:是|为|约为|目标为)?\s*({_NUMBER})\s*%", + text, + re.IGNORECASE, + ) + if single_return: + value = _number(single_return.group(1)) + result["annualized_return_lower_pct"] = value + result["annualized_return_upper_pct"] = value + + drawdown = re.search(rf"(?:最大)?回撤\s*(?:容忍|控制|不超过|为|约为)?\s*({_NUMBER})\s*%?", text) + if drawdown: + result["max_drawdown_pct"] = _number(drawdown.group(1)) + + if re.search(r"随时|每天|每日|当天|日内", text): + result["liquidity_requirement"] = "daily" + else: + days = re.search( + r"(?:流动性|用钱|变现|取出).{0,12}?" + r"([一两两三四五六七八九十0-9]+)\s*天", + text, + ) + if days is None: + days = re.search( + r"([一两三四五六七八九十0-9]+)\s*天(?:内|可用|能用)", text + ) + if days: + count = _chinese_number(days.group(1)) + if count <= 7: + result["liquidity_requirement"] = "within_7_days" + elif count <= 30: + result["liquidity_requirement"] = "within_30_days" + else: + result["liquidity_requirement"] = "over_30_days" + + months = re.search(rf"({_NUMBER})\s*(?:年|岁)", text) + if months: + result["investment_horizon_months"] = int(_number(months.group(1)) * 12) + else: + months = re.search(rf"({_NUMBER})\s*个?月", text) + if months: + result["investment_horizon_months"] = int(_number(months.group(1))) + + benchmark = re.search( + r"(?:业绩比较基准|比较基准|基准)\s*(?:是|为|选)?\s*" + r"([\u4e00-\u9fffA-Za-z0-9_-]{2,32})", + text, + ) + if benchmark: + result["benchmark_name"] = benchmark.group(1).strip("，。；、") + else: + for candidate in ("沪深300指数", "沪深300", "中证500", "中证全指", "上证指数"): + if candidate in text: + result["benchmark_name"] = candidate + break + return result + + +def _chinese_number(value: str) -> int: + if value.isdigit(): + return int(value) + digits = {"一": 1, "二": 2, "两": 2, "三": 3, "四": 4, "五": 5, + "六": 6, "七": 7, "八": 8, "九": 9, "十": 10} + if value == "十": + return 10 + if value.startswith("十"): + return 10 + digits.get(value[1:], 0) + if value.endswith("十"): + return digits.get(value[0], 0) * 10 + if "十" in value: + left, right = value.split("十", 1) + return digits.get(left, 0) * 10 + digits.get(right, 0) + return digits.get(value, 0) + + +class GoalConversationService: + """Merge explicit facts across a session and persist an internal gap snapshot.""" + + def __init__( + self, session_factory: SessionFactoryLike = SessionFactory, + ) -> None: + self.session_factory = session_factory + + async def process( + self, *, session_id: str, customer_id: int, trace_id: str, message: str + ) -> dict[str, Any]: + async with self.session_factory() as session, session.begin(): + session_row = await session.scalar(select(ConversationSession).where( + ConversationSession.session_id == session_id, + ConversationSession.user_id == customer_id, + ).with_for_update()) + if session_row is None: + return self._result({}, list(REQUIRED_FIELDS), "partial") + messages = list(await session.scalars(select(ConversationMessage).where( + ConversationMessage.session_id == session_id, + ConversationMessage.customer_id == customer_id, + ConversationMessage.role == "user", + ).order_by(ConversationMessage.created_at, ConversationMessage.id))) + merged: dict[str, Any] = {} + for row in messages: + merged.update(extract_goal_entities(row.content)) + merged.update(extract_goal_entities(message)) + missing = [field for field in REQUIRED_FIELDS if field not in merged] + status = "complete" if not missing else ( + "clarification_limit" + if session_row.clarification_round >= MAX_CLARIFICATION_ROUNDS + else "partial" + ) + if status == "partial": + session_row.clarification_round += 1 + now = datetime.now(UTC).replace(tzinfo=None) + message_row = await session.scalar(select(ConversationMessage).where( + ConversationMessage.session_id == session_id, + ConversationMessage.customer_id == customer_id, + ConversationMessage.trace_id == trace_id, + ConversationMessage.role == "user", + ).order_by(ConversationMessage.id.desc()).limit(1)) + if message_row is not None: + exists = await session.scalar(select(AdvisorGoalConversationExtraction).where( + AdvisorGoalConversationExtraction.message_id == message_row.id)) + if exists is None: + session.add(AdvisorGoalConversationExtraction( + message_id=message_row.id, session_id=session_id, + customer_id=customer_id, extraction_version=EXTRACTION_VERSION, + extracted_fields=merged, missing_fields=missing, status=status, + created_at=now, updated_at=now, + )) + session.add(InteractionAudit( + actor_type="agent", actor_id=customer_id, session_id=session_id, + portal="agent", action_type="advisor.goal_gap_evaluated", + detail={"trace_id": trace_id, "status": status, + "missing_count": len(missing)}, created_at=now, + )) + return self._result(merged, missing, status) + + @staticmethod + def _result(fields: dict[str, Any], missing: list[str], status: str) -> dict[str, Any]: + return {"fields": fields, "missing": missing, "status": status} + + @staticmethod + def customer_prompt(result: dict[str, Any]) -> str: + missing = result.get("missing", []) + if result.get("status") == "clarification_limit": + return "投资目标信息仍不完整，暂时无法继续采集，请转人工顾问协助确认。" + if missing: + labels = [FIELD_LABELS[field] for field in missing if field in FIELD_LABELS] + return "为了形成投资目标，还需要确认：" + "、".join(labels) + "。请一次补充这些信息。" + fields = result["fields"] + return ( + f"投资目标信息已收集完整：年化收益目标 {fields['annualized_return_lower_pct']}%-" + f"{fields['annualized_return_upper_pct']}%，最大回撤 {fields['max_drawdown_pct']}%，" + f"流动性 {fields['liquidity_requirement']}，" + f"期限 {fields['investment_horizon_months']} 个月，" + f"业绩比较基准 {fields['benchmark_name']}。请确认后再生成目标书。" + ) + + @staticmethod + def should_collect(text: str) -> bool: + """Distinguish goal collection from a request to view an existing goal.""" + return bool(extract_goal_entities(text)) or bool(re.search( + r"制定|设定|创建|填写|采集|规划|目标书", text + )) diff --git a/docs/21-投顾Agent迁移TODO.md b/docs/21-投顾Agent迁移TODO.md index 2ff7a82..4b038c1 100644 --- a/docs/21-投顾Agent迁移TODO.md +++ b/docs/21-投顾Agent迁移TODO.md @@ -111,6 +111,19 @@ Agent 暴露客户最新的 `confirmed` 目标，未确认目标不会进入后 阶段十测试结果：推荐服务专项测试 `3 passed`，全量单元/契约测试 `493 passed, 3 warnings`， Ruff 通过，MyPy（138 个源文件）通过。阶段十提交：`0d42779`；生产数据库和端到端联调待完成。 +阶段十一已完成会话闭环核心：投顾 Agent 支持投资目标、产品推荐、持仓分析、资产配置、 +行情和对比意图路由；投资目标会话抽取只接受用户明确陈述，按会话合并多轮字段，写入内部 +`advisor_goal_conversation_extraction` 快照并根据缺口追问。缺口状态使用既有 +`svc_conversation_session.clarification_round` 持久化，达到 10 轮后转人工；Agent Run 完成 +时同步保存 `last_intent`。目标抽取记录、会话审计、Agent Run、Outbox、记忆召回和 Episode +聚合均沿用公共底座链路，客户响应不返回抽取字段、置信度、内部表名或治理信息。对比意图当前 +完成安全路由和参数提示，对比计算工具仍是后续待办。 + +阶段十一测试结果：专项 `4 passed`；单元/契约 `497 passed, 3 warnings`；Ruff 通过；MyPy +（140 个源文件）通过。集成测试 `25 passed, 3 failed, 1 skipped`：失败均为既有测试库状态 +问题（`chk_config_release_separation` 未按新底座迁移撤下、测试账号外键缺失，以及 UTC +测试依赖的数据库状态），不是本阶段代码回归；独立迁移库和端到端会话验收仍待执行。 + ## 一、迁移准备 - [ ] 确认远程仓库可访问。（当前失败：连接 `47.106.207.27:3000` 被拒绝） @@ -369,25 +382,25 @@ python tools/audit_constraints.py ## 十一、会话闭环 -- [ ] 迁移意图分类。 -- [ ] 迁移产品推荐意图。 -- [ ] 迁移持仓分析意图。 -- [ ] 迁移资产配置意图。 -- [ ] 迁移对比分析意图。 -- [ ] 迁移投资目标意图。 -- [ ] 迁移会话实体抽取。 -- [ ] 迁移投资目标缺口识别。 -- [ ] 迁移缺口追问状态。 -- [ ] 迁移会话状态持久化。 -- [ ] 迁移记忆召回。 -- [ ] 迁移 Episode 聚合。 -- [ ] 迁移 Agent Run 持久化。 -- [ ] 迁移 Outbox 事件。 -- [ ] 迁移会话审计。 -- [ ] 确认用户消息可以形成完整闭环。 -- [ ] 确认缺少字段时只追问必要信息。 -- [ ] 确认工具失败时返回可理解的降级结果。 -- [ ] 完成会话闭环提交 `advisor/conversation`。 +- [x] 迁移意图分类。（沿用公共 `IntentClassifier`；新增对比意图声明） +- [x] 迁移产品推荐意图。 +- [x] 迁移持仓分析意图。 +- [x] 迁移资产配置意图。 +- [x] 迁移对比分析意图。（已完成安全路由和参数提示；计算工具待补） +- [x] 迁移投资目标意图。 +- [x] 迁移会话实体抽取。（`GoalConversationService`，只抽取明确目标字段） +- [x] 迁移投资目标缺口识别。（按当前会话所有用户轮次合并） +- [x] 迁移缺口追问状态。（使用 `clarification_round`，上限 10 轮） +- [x] 迁移会话状态持久化。（保存 `last_intent` 和缺口轮次） +- [x] 迁移记忆召回。（复用 `PlatformGovernance` 组合召回） +- [x] 迁移 Episode 聚合。（复用 `EpisodeWorker`） +- [x] 迁移 Agent Run 持久化。（复用 `AgentPersistenceService`） +- [x] 迁移 Outbox 事件。（复用 Agent Run 和记忆事件链路） +- [x] 迁移会话审计。（目标缺口评估和运行/工具均写统一审计） +- [x] 确认用户消息可以形成完整闭环。（专项 + 单元/契约测试通过） +- [x] 确认缺少字段时只追问必要信息。（专项测试覆盖） +- [x] 确认工具失败时返回可理解的降级结果。（既有各业务工具降级契约通过） +- [x] 完成会话闭环提交 `advisor/conversation`。（真实库端到端验收待完成） ## 十二、画像标签和漂移复核 diff --git a/tests/unit/service/test_advisor_base_adapter.py b/tests/unit/service/test_advisor_base_adapter.py index 0e441e6..54e7622 100644 --- a/tests/unit/service/test_advisor_base_adapter.py +++ b/tests/unit/service/test_advisor_base_adapter.py @@ -31,4 +31,5 @@ def test_advisor_is_registered_through_the_new_base_factory() -> None: "portfolio_analysis", "asset_allocation", "product_recommend", + "comparison", ) diff --git a/tests/unit/service/test_goal_conversation_service.py b/tests/unit/service/test_goal_conversation_service.py new file mode 100644 index 0000000..fe656a1 --- /dev/null +++ b/tests/unit/service/test_goal_conversation_service.py @@ -0,0 +1,142 @@ +from typing import Any + +import pytest + +from app.service.goal_conversation_service import ( + MAX_CLARIFICATION_ROUNDS, + REQUIRED_FIELDS, + GoalConversationService, + extract_goal_entities, +) + + +def test_extracts_explicit_goal_entities() -> None: + result = extract_goal_entities( + "我希望年化收益6%-10%，最大回撤15%，7天内能用钱，期限3年，基准沪深300指数" + ) + + assert result == { + "annualized_return_lower_pct": 6.0, + "annualized_return_upper_pct": 10.0, + "max_drawdown_pct": 15.0, + "liquidity_requirement": "within_7_days", + "investment_horizon_months": 36, + "benchmark_name": "沪深300指数", + } + + +def test_collection_detection_does_not_intercept_goal_read() -> None: + assert GoalConversationService.should_collect("我想制定投资目标") + assert GoalConversationService.should_collect("收益目标是8%") + assert not GoalConversationService.should_collect("查看我的投资目标") + + +class _AsyncContext: + async def __aenter__(self) -> None: + return None + + async def __aexit__(self, *_args: object) -> bool: + return False + + +class _SessionRow: + clarification_round = 0 + + +class _Message: + def __init__(self, message_id: int, content: str, trace_id: str) -> None: + self.id = message_id + self.content = content + self.trace_id = trace_id + + +class _FakeSession: + def __init__(self, messages: list[_Message], rounds: int = 0) -> None: + self.session_row = _SessionRow() + self.session_row.clarification_round = rounds + self.messages = messages + self.added: list[Any] = [] + self.scalar_calls = 0 + + def begin(self) -> _AsyncContext: + return _AsyncContext() + + async def scalar(self, _statement: object) -> object: + self.scalar_calls += 1 + if self.scalar_calls == 1: + return self.session_row + if self.scalar_calls == 2: + return next((item for item in self.messages if item.trace_id == "trace-2"), None) + return None + + async def scalars(self, _statement: object) -> list[_Message]: + return self.messages + + def add(self, item: Any) -> None: + self.added.append(item) + + +class _SessionFactory: + def __init__(self, session: _FakeSession) -> None: + self.session = session + + def __call__(self) -> "_SessionContext": + return _SessionContext(self.session) + + +class _SessionContext: + def __init__(self, session: _FakeSession) -> None: + self.session = session + + async def __aenter__(self) -> _FakeSession: + return self.session + + async def __aexit__(self, *_args: object) -> bool: + return False + + +@pytest.mark.asyncio +async def test_process_merges_turns_and_persists_only_internal_snapshot() -> None: + session = _FakeSession([ + _Message(1, "我想做投资目标，收益6%-10%，最大回撤15%", "trace-1"), + _Message(2, "7天内用钱，期限3年，基准沪深300", "trace-2"), + ]) + + result = await GoalConversationService(_SessionFactory(session)).process( + session_id="session-1", customer_id=9000001, trace_id="trace-2", + message=session.messages[-1].content, + ) + + assert result["status"] == "complete" + assert result["missing"] == [] + assert session.session_row.clarification_round == 0 + extraction = next( + item for item in session.added + if item.__class__.__name__ == "AdvisorGoalConversationExtraction" + ) + assert extraction.missing_fields == [] + assert extraction.extracted_fields["investment_horizon_months"] == 36 + + +@pytest.mark.asyncio +async def test_process_asks_only_for_missing_fields_and_honors_limit() -> None: + session = _FakeSession([_Message(3, "我希望年化收益6%-10%", "trace-2")]) + + result = await GoalConversationService(_SessionFactory(session)).process( + session_id="session-1", customer_id=9000001, trace_id="trace-2", + message="我希望年化收益6%-10%", + ) + + assert result["status"] == "partial" + assert result["missing"] == [field for field in REQUIRED_FIELDS if field not in { + "annualized_return_lower_pct", "annualized_return_upper_pct" + }] + assert session.session_row.clarification_round == 1 + assert "年化收益目标下限" not in GoalConversationService.customer_prompt(result) + + limited = _FakeSession(session.messages, rounds=MAX_CLARIFICATION_ROUNDS) + limited_result = await GoalConversationService(_SessionFactory(limited)).process( + session_id="session-1", customer_id=9000001, trace_id="trace-2", message="我还没想好" + ) + assert limited_result["status"] == "clarification_limit" + assert "转人工顾问" in GoalConversationService.customer_prompt(limited_result)