From 62bf352007a8bab6bf64a3f779aac260a1bbfa38 Mon Sep 17 00:00:00 2001 From: TNT-Likely Date: Tue, 29 Sep 2026 07:28:51 +0800 Subject: [PATCH] feat(assistant): add trusted results and actionable research --- frontend/packages/api/src/chat.ts | 77 ++- frontend/src/components/ChatWidget.tsx | 18 + .../assistant/AssistantResultCard.tsx | 239 +++++++ .../components/assistant/TraceTimeline.tsx | 70 +- .../src/i18n/locales/en-US/configuration.ts | 6 +- .../src/i18n/locales/zh-CN/configuration.ts | 6 +- frontend/src/lib/assistant-navigation.ts | 20 + frontend/src/pages/Assistant.tsx | 1 + frontend/src/pages/Stocks.tsx | 21 + frontend/tests/api/chatStream.test.ts | 33 + .../assistant/AssistantResultCard.test.tsx | 79 +++ .../tests/assistant/ChatWidgetLayout.test.tsx | 6 +- .../tests/assistant/TraceTimeline.test.tsx | 20 +- .../assistant/assistant-navigation.test.ts | 19 + .../src/pan_agent/contracts.py | 4 + .../src/pan_agent/runtime.py | 38 +- .../pan-agent-runtime/tests/test_runtime.py | 18 +- src/modules/assistant/api.py | 2 + src/modules/assistant/event_stream.py | 24 +- src/modules/assistant/prompt.py | 4 + src/modules/assistant/repository.py | 126 +++- src/modules/assistant/result_builder.py | 631 ++++++++++++++++++ src/modules/assistant/result_schemas.py | 48 ++ src/modules/assistant/schemas.py | 6 +- src/modules/assistant/service.py | 101 ++- src/modules/assistant/task_runner.py | 85 ++- src/modules/assistant/tools.py | 70 +- src/modules/assistant/trace.py | 62 ++ .../marketdata/collectors/kline_collector.py | 2 + src/platform/persistence/migrations.py | 33 + src/platform/persistence/models.py | 5 + tests/test_assistant_results.py | 169 +++++ tests/test_assistant_stream_api.py | 11 +- tests/test_assistant_task_events.py | 31 + tests/test_assistant_task_repository.py | 2 + 35 files changed, 1994 insertions(+), 93 deletions(-) create mode 100644 frontend/src/components/assistant/AssistantResultCard.tsx create mode 100644 frontend/src/lib/assistant-navigation.ts create mode 100644 frontend/tests/assistant/AssistantResultCard.test.tsx create mode 100644 frontend/tests/assistant/assistant-navigation.test.ts create mode 100644 src/modules/assistant/result_builder.py create mode 100644 src/modules/assistant/result_schemas.py create mode 100644 src/modules/assistant/trace.py create mode 100644 tests/test_assistant_results.py diff --git a/frontend/packages/api/src/chat.ts b/frontend/packages/api/src/chat.ts index 83c57b94..2621d434 100644 --- a/frontend/packages/api/src/chat.ts +++ b/frontend/packages/api/src/chat.ts @@ -17,6 +17,56 @@ export interface ChatMessage { created_at: string /** Runtime facts captured while producing this assistant response. */ trace?: AssistantTraceEvent[] + result?: AssistantResult | null +} + +export interface AssistantEvidence { + id: string + tool_name: string + source_name: string + source_url?: string | null + summary: string + observed_at?: string | null + data_at?: string | null + period_start?: string | null + period_end?: string | null + freshness: 'fresh' | 'delayed' | 'stale' | 'unknown' + freshness_basis: 'published_at' | 'as_of' | 'observed_at' | 'unknown' + symbol?: string | null + market?: string | null +} + +export interface AssistantSource { + name: string + url?: string | null + as_of?: string | null + published_at?: string | null + period_start?: string | null + period_end?: string | null +} + +export interface AssistantFact { + text: string + evidence_ids: string[] +} + +export interface AssistantNextAction { + id: string + kind: 'follow_up' | 'navigate' | 'tool_proposal' + label: string + payload: Record + requires_approval: boolean +} + +export interface AssistantResult { + schema_version: number + summary: string + facts: AssistantFact[] + inferences: string[] + risks: string[] + missing_data: string[] + evidence: AssistantEvidence[] + next_actions: AssistantNextAction[] } export interface ConversationDetail { @@ -57,7 +107,10 @@ export interface AssistantTaskSnapshot { duration_ms: number attempt_count: number error_code?: string | null + sources?: AssistantSource[] + observed_at?: string | null }> + result?: AssistantResult | null pending_approvals: Array<{ id: string call_id: string @@ -241,7 +294,13 @@ export interface ChatStreamCallbacks { /** 模型开始调用工具（前端应清空当前 token 缓冲并展示"正在查询…"） */ onToolCallStart?: (info: { name: string; arguments: Record }) => void /** 工具执行完成 */ - onToolResult?: (info: { name: string; ok: boolean; preview: string }) => void + onToolResult?: (info: { + name: string + ok: boolean + preview: string + sources: AssistantSource[] + observedAt?: string | null + }) => void /** 计划驱动(全面诊断持仓):计划生成/步骤推进/完成 */ onPlan?: (info: { status: string @@ -258,7 +317,7 @@ export interface ChatStreamCallbacks { resolvedStatus?: AssistantApproval['status'] }) => void /** 最终回答（已落库） */ - onDone?: (msg: { message_id: number; content: string; created_at: string }) => void + onDone?: (msg: { message_id: number; content: string; created_at: string; result?: AssistantResult | null }) => void /** AI service failure with a safe stable code and localized fallback text. */ onError?: (error: AssistantStreamError) => void /** Factual runtime events for the user-facing trace panel. */ @@ -308,7 +367,10 @@ function dispatchAssistantEvent( if (ev.id > 0) state.lastEventId = ev.id const d = ev.data || {} if (TRACE_EVENTS.has(ev.event)) { - callbacks.onTrace?.({ event: ev.event, data: d, id: ev.id }) + const traceData = ev.event === 'done' + ? Object.fromEntries(Object.entries(d).filter(([key]) => !['content', 'result'].includes(key))) + : d + callbacks.onTrace?.({ event: ev.event, data: traceData, id: ev.id }) } switch (ev.event) { case 'task_created': @@ -355,7 +417,13 @@ function dispatchAssistantEvent( callbacks.onToolCallStart?.({ name: d.name || d.tool || '', arguments: d.arguments || {} }) break case 'tool_result': - callbacks.onToolResult?.({ name: d.name || d.tool || '', ok: !!d.ok, preview: d.preview || d.summary || '' }) + callbacks.onToolResult?.({ + name: d.name || d.tool || '', + ok: !!d.ok, + preview: d.preview || d.summary || '', + sources: Array.isArray(d.sources) ? d.sources : [], + observedAt: d.observed_at || null, + }) break case 'plan': callbacks.onPlan?.({ status: d.status || '', steps: d.steps || [], current: d.current }) @@ -389,6 +457,7 @@ function dispatchAssistantEvent( message_id: d.message_id || 0, content: d.content || '', created_at: d.created_at || '', + result: d.result || null, }) break case 'error': diff --git a/frontend/src/components/ChatWidget.tsx b/frontend/src/components/ChatWidget.tsx index 127f54b2..4fc1d1f0 100644 --- a/frontend/src/components/ChatWidget.tsx +++ b/frontend/src/components/ChatWidget.tsx @@ -22,6 +22,7 @@ import { ContextUsageIndicator } from '@/components/assistant/ContextUsageIndica import { TraceTimeline } from '@/components/assistant/TraceTimeline' import { useChatAutoScroll } from '@/hooks/useChatAutoScroll' import { useTranslation } from 'react-i18next' +import { AssistantResultCard } from '@/components/assistant/AssistantResultCard' interface StockContext { symbol: string @@ -42,6 +43,8 @@ interface ChatWidgetProps { onConversationChange?: (conversationId: number | null, options?: ConversationChangeOptions) => void /** Stock context handed off by the application shell when a page opens “问 AI”. */ initialStockContext?: StockContext | null + /** Navigate to a trusted application route selected from a result action. */ + onNavigate?: (path: string) => void } function taskStorageKey(conversationId: number): string { @@ -98,6 +101,7 @@ export default function ChatWidget({ conversationIdFromUrl = null, onConversationChange, initialStockContext = null, + onNavigate, }: ChatWidgetProps) { const { t } = useTranslation('configuration') const assistantT = t as unknown as (key: string, options?: Record) => string @@ -730,6 +734,7 @@ export default function ChatWidget({ content: m.content, created_at: m.created_at || new Date().toISOString(), trace: completedTrace.length > 0 ? completedTrace : undefined, + result: m.result, }]) traceEventsRef.current = [] setTraceEvents([]) @@ -840,6 +845,7 @@ export default function ChatWidget({ content: message.content, created_at: message.created_at || new Date().toISOString(), trace: completedTrace.length > 0 ? completedTrace : undefined, + result: message.result, }]) traceEventsRef.current = [] setTraceEvents([]) @@ -1142,6 +1148,18 @@ export default function ChatWidget({ msg.content )} + {msg.role === 'assistant' && msg.result && ( + { + setInput(prompt) + requestAnimationFrame(() => inputRef.current?.focus()) + }} + onNavigate={(path) => onNavigate?.(path)} + onSubmitPrompt={(prompt) => { void handleSend(prompt) }} + /> + )} {msg.role === 'assistant' && msg.trace && msg.trace.length > 0 && ( )} diff --git a/frontend/src/components/assistant/AssistantResultCard.tsx b/frontend/src/components/assistant/AssistantResultCard.tsx new file mode 100644 index 00000000..8ddd922d --- /dev/null +++ b/frontend/src/components/assistant/AssistantResultCard.tsx @@ -0,0 +1,239 @@ +import { useMemo, useState } from 'react' +import { Bell, ChevronDown, ExternalLink, FileSearch, ShieldAlert, Sparkles } from 'lucide-react' +import { useTranslation } from 'react-i18next' +import type { AssistantNextAction, AssistantResult } from '@panwatch/api' + +interface AssistantResultCardProps { + result: AssistantResult + disabled?: boolean + onPrefill: (prompt: string) => void + onNavigate: (path: string) => void + onSubmitPrompt: (prompt: string) => void +} + +const ALLOWED_PATHS = new Set(['/portfolio', '/opportunities', '/alerts']) + +function safeInternalPath(value: string): string | null { + try { + const url = new URL(value, window.location.origin) + if (url.origin !== window.location.origin || !ALLOWED_PATHS.has(url.pathname)) return null + return `${url.pathname}${url.search}${url.hash}` + } catch { + return null + } +} + +function formatObservedAt(value: string | null | undefined, locale: string): string { + if (!value) return '' + const date = new Date(value) + if (Number.isNaN(date.getTime())) return '' + return new Intl.DateTimeFormat(locale, { + month: 'numeric', day: 'numeric', hour: '2-digit', minute: '2-digit', + }).format(date) +} + +function safeExternalUrl(value: string | null | undefined): string | null { + if (!value) return null + try { + const url = new URL(value) + return url.protocol === 'http:' || url.protocol === 'https:' ? url.toString() : null + } catch { + return null + } +} + +export function AssistantResultCard({ + result, + disabled = false, + onPrefill, + onNavigate, + onSubmitPrompt, +}: AssistantResultCardProps) { + const { t, i18n } = useTranslation('configuration') + const resultT = t as unknown as (key: string, options?: Record) => string + const tr = (key: string, options?: Record) => resultT(`assistantPage.result.${key}`, options) + const [evidenceOpen, setEvidenceOpen] = useState(false) + const [alertAction, setAlertAction] = useState(null) + const [direction, setDirection] = useState<'above' | 'below'>('below') + const [targetPrice, setTargetPrice] = useState('') + const evidenceById = useMemo( + () => new Map(result.evidence.map((item) => [item.id, item])), + [result.evidence], + ) + const hasDetails = result.facts.length > 0 + || result.inferences.length > 0 + || result.risks.length > 0 + || result.missing_data.length > 0 + + const handleAction = (action: AssistantNextAction) => { + if (disabled) return + if (action.kind === 'follow_up' && typeof action.payload.prompt === 'string') { + onPrefill(action.payload.prompt) + return + } + if (action.kind === 'navigate' && typeof action.payload.path === 'string') { + const path = safeInternalPath(action.payload.path) + if (path) { + onNavigate(path) + return + } + } + if (action.kind === 'tool_proposal' && action.payload.tool_name === 'create_price_alert') { + setAlertAction(action) + } + } + + const submitAlert = () => { + if (!alertAction || disabled) return + const price = Number(targetPrice) + if (!Number.isFinite(price) || price <= 0) return + const args = alertAction.payload.arguments || {} + const market = String(args.market || 'CN').toUpperCase() + const symbol = String(args.symbol || '').toUpperCase() + if (!symbol) return + const prompt = tr('alertPrompt', { + target: `${market}:${symbol}`, + direction: direction === 'above' ? tr('above') : tr('below'), + price: String(price), + }) + onSubmitPrompt(prompt) + setAlertAction(null) + setTargetPrice('') + } + + if (!hasDetails && result.next_actions.length === 0) return null + + return ( +
+ {result.facts.length > 0 && ( +
+
+ + {tr('facts')} +
+
    + {result.facts.slice(0, 4).map((fact, index) => { + const sources = fact.evidence_ids + .map((id) => evidenceById.get(id)) + .filter(Boolean) + return ( +
  • + •{fact.text} + {sources.length > 0 && ( + + [{sources.map((item) => item?.source_name).filter(Boolean).join(' · ')}] + + )} +
    • + ) + })} +
+
+ )} + + {(result.inferences.length > 0 || result.risks.length > 0 || result.missing_data.length > 0) && ( +
+ {result.inferences.length > 0 && ( +
+
{tr('inferences')}
+ {result.inferences.slice(0, 3).map((item, index) =>
• {item}
)} +
+ )} + {(result.risks.length > 0 || result.missing_data.length > 0) && ( +
+
{tr('risks')}
+ {[...result.risks, ...result.missing_data].slice(0, 3).map((item, index) =>
• {item}
)} +
+ )} +
+ )} + + {result.evidence.length > 0 && ( +
+ + {evidenceOpen && ( +
    + {result.evidence.map((item) => { + const sourceUrl = safeExternalUrl(item.source_url) + return ( +
  • +
    + {sourceUrl ? ( + + {item.source_name} + + ) : {item.source_name}} + + {tr(`freshness.${item.freshness}`)} + + {item.observed_at && {tr('observedAt', { time: formatObservedAt(item.observed_at, i18n.language) })}} + {item.data_at && {tr('dataAt', { time: formatObservedAt(item.data_at, i18n.language) || item.data_at })}} + {(item.period_start || item.period_end) && ( + + {tr('coverage', { start: item.period_start || '—', end: item.period_end || '—' })} + + )} +
    +
    {item.summary}
    +
    • + ) + })} +
+ )} +
+ )} + + {result.next_actions.length > 0 && ( +
+
{tr('nextActions')}
+
+ {result.next_actions.map((action) => ( + + ))} +
+
+ )} + + {alertAction && ( +
+
{tr('configureAlert')}
+
+ + setTargetPrice(event.target.value)} + placeholder={tr('targetPrice')} + className="h-8 min-w-0 flex-1 rounded-md border border-border bg-background px-2 text-[11px]" + /> + +
+
+ )} +
+ ) +} diff --git a/frontend/src/components/assistant/TraceTimeline.tsx b/frontend/src/components/assistant/TraceTimeline.tsx index a6cbef85..10316ed5 100644 --- a/frontend/src/components/assistant/TraceTimeline.tsx +++ b/frontend/src/components/assistant/TraceTimeline.tsx @@ -27,14 +27,15 @@ function describeExtensionEvent(event: AssistantTraceEvent, tr: (key: string, op function describe(event: AssistantTraceEvent, tr: (key: string, options?: Record) => string): { label: string; icon: typeof FileClock } { const name = typeof event.data.name === 'string' ? event.data.name : '' + const displayName = name ? tr(`tools.${name}`, { defaultValue: name }) : name const extension = event.event === 'extension_event' ? describeExtensionEvent(event, tr) : null if (extension) return extension switch (event.event) { case 'context_prepared': return { label: tr(event.data.compressed ? 'events.contextCompressed' : 'events.contextPrepared'), icon: FileClock } case 'step_updated': return { label: tr('events.step', { step: event.data.step || '' }), icon: ListTree } - case 'tool_call_start': return { label: tr('events.toolStart', { name }), icon: Wrench } + case 'tool_call_start': return { label: tr('events.toolStart', { name: displayName }), icon: Wrench } case 'tool_result': return { - label: `${tr(event.data.ok ? 'events.toolDone' : 'events.toolFailed', { name })}${formatDurationSuffix(event.data.duration_ms)}`, + label: `${tr(event.data.ok ? 'events.toolDone' : 'events.toolFailed', { name: displayName })}${formatDurationSuffix(event.data.duration_ms)}`, icon: event.data.ok ? CheckCircle2 : AlertCircle, } case 'model_usage': { @@ -81,13 +82,6 @@ function detail(event: AssistantTraceEvent): string { } function summary(events: AssistantTraceEvent[], tr: (key: string, options?: Record) => string): string { - const toolCalls = events.filter((event) => event.event === 'tool_call_start').length - const totalTokens = events - .filter((event) => event.event === 'model_usage') - .reduce((total, event) => total + Number( - event.data.total_tokens - || Number(event.data.input_tokens || 0) + Number(event.data.output_tokens || 0), - ), 0) const terminal = [...events].reverse().find((event) => ['done', 'error', 'paused'].includes(event.event)) const duration = Number(terminal?.data.duration_ms || 0) const status = terminal?.event === 'done' @@ -97,12 +91,26 @@ function summary(events: AssistantTraceEvent[], tr: (key: string, options?: Reco : terminal?.event === 'paused' ? tr('status.paused') : tr('status.running') - const details = [ - formatDuration(duration), + const details = [formatDuration(duration)].filter(Boolean) + return details.length > 0 ? `${status} · ${details.join(' · ')}` : status +} + +function developerSummary(events: AssistantTraceEvent[], tr: (key: string, options?: Record) => string): string { + const toolCalls = events.filter((event) => event.event === 'tool_call_start').length + const totalTokens = events + .filter((event) => event.event === 'model_usage') + .reduce((total, event) => total + Number( + event.data.total_tokens + || Number(event.data.input_tokens || 0) + Number(event.data.output_tokens || 0), + ), 0) + return [ toolCalls > 0 ? tr('toolCalls', { count: toolCalls }) : '', totalTokens > 0 ? `${totalTokens} tokens` : '', - ].filter(Boolean) - return details.length > 0 ? `${status} · ${details.join(' · ')}` : status + ].filter(Boolean).join(' · ') +} + +function userVisible(event: AssistantTraceEvent): boolean { + return !['model_usage', 'extension_event'].includes(event.event) } export function TraceTimeline({ events, live = false }: TraceTimelineProps) { @@ -110,6 +118,8 @@ export function TraceTimeline({ events, live = false }: TraceTimelineProps) { const traceT = t as unknown as (key: string, options?: Record) => string const tr = (key: string, options?: Record) => traceT(`p4.components.trace.${key}`, options) const [expanded, setExpanded] = useState(live) + const [developerExpanded, setDeveloperExpanded] = useState(false) + const userEvents = events.filter(userVisible) if (events.length === 0) return null return (
@@ -126,19 +136,47 @@ export function TraceTimeline({ events, live = false }: TraceTimelineProps) { {expanded && (
    - {events.map((event, index) => { + {userEvents.map((event, index) => { const { label, icon: Icon } = describe(event, tr) - const eventDetail = detail(event) return (
  1. {label}
    - {eventDetail &&
    {eventDetail}
    }
    2. ) })} +
  2. + + {developerExpanded && ( +
      + {events.map((event, index) => { + const { label, icon: Icon } = describe(event, tr) + const eventDetail = detail(event) + return ( +
    1. + +
      +
      {label}
      + {eventDetail &&
      {eventDetail}
      } +
      +
      2. + ) + })} +
    + )} +
)}
diff --git a/frontend/src/i18n/locales/en-US/configuration.ts b/frontend/src/i18n/locales/en-US/configuration.ts index 214b3508..b92b8655 100644 --- a/frontend/src/i18n/locales/en-US/configuration.ts +++ b/frontend/src/i18n/locales/en-US/configuration.ts @@ -323,7 +323,7 @@ export const configuration = { permissionsDrawer: { loadFailed: 'Unable to load tool permissions. Try again later.', saveFailed: 'Failed to save permissions. Try again.', title: 'Assistant settings', description: 'Manage tool permissions and the model and budget used for context compression.', loading: 'Loading tool permissions…' }, contextPanel: { sections: { system: 'System instructions', summary: 'Structured summary', page_context: 'Page context', tool_definitions: 'Tool definitions', history: 'Conversation history', recent_messages: 'Recent messages' }, statuses: { not_needed: 'Compression was not needed', compressed: 'Compression complete', no_gain: 'Compression did not reduce context; the original was preserved' }, modes: { balanced: 'Balanced compression', preserve_details: 'Compress while preserving details', handoff: 'Create a handoff summary' }, measurement: { provider: 'Actual input tokens', tokenizer: 'Tokenizer input tokens', estimate: 'Estimated input tokens' }, title: 'Context usage', description: 'Applies to the next run only; conversation messages are preserved.', close: 'Close context panel', measuring: 'Measuring context…', needsCompression: 'Needs compression', warning: 'Near the limit', normal: 'Normal', snapshot: 'Summary v{{version}}', goal: 'Goal: {{value}}', state: 'State: {{value}}', todo: 'To do: {{value}}', savedTokens: '{{before}} → {{after}}; saved {{saved}} tokens ({{percent}}%)', compressing: 'Compressing…', empty: 'No measurable conversation yet.' }, contextUsage: { states: { normal: 'Context is healthy', warning: 'Context is large', needs_compression: 'Compression needed' }, aria: 'View context usage', title: '{{state}}; estimated input tokens {{used}} / {{budget}}' }, - trace: { extension: { started: 'Finding available tools', exposure: 'Tool catalog ready: {{direct}} direct, {{loaded}} loaded', candidates: 'Scored {{count}} tool candidates', completed: 'Tool research complete: selected {{count}}', searched: 'Tool search complete: loaded {{count}}', fallback: 'Tool research fallback; continuing with the default tool set', unknown: 'Extension event: {{event}}' }, events: { contextCompressed: 'Context compressed and prepared', contextPrepared: 'Context prepared', step: 'Step {{step}}', toolStart: 'Calling tool: {{name}}', toolDone: 'Tool completed: {{name}}', toolFailed: 'Tool failed: {{name}}', modelUsage: 'Model usage: input {{input}}, output {{output}}{{extra}}', cache: 'Cached {{count}}', reasoning: 'Reasoning {{count}}', approval: 'Waiting for approval', paused: 'Task paused', done: 'Task completed', error: 'Task failed', started: 'Task started', title: 'Execution trace' }, status: { done: 'Completed', error: 'Failed', paused: 'Waiting to resume', running: 'Running' }, toolCalls: '{{count}} tool calls' }, + trace: { extension: { started: 'Finding available tools', exposure: 'Tool catalog ready: {{direct}} direct, {{loaded}} loaded', candidates: 'Scored {{count}} tool candidates', completed: 'Tool research complete: selected {{count}}', searched: 'Tool search complete: loaded {{count}}', fallback: 'Tool research fallback; continuing with the default tool set', unknown: 'Extension event: {{event}}' }, tools: { get_portfolio: 'portfolio', get_stock_quote: 'live quote', find_research_candidates: 'research candidates', get_kline_summary: 'chart trend', get_stock_news: 'recent news', search_stocks: 'stock search', get_market_status: 'market status', get_hot_stocks: 'hot stocks', get_hot_boards: 'hot sectors', get_board_stocks: 'sector constituents', get_stock_fundamentals: 'fundamentals', get_capital_flow: 'capital flow', get_dragon_tiger: 'Dragon Tiger List', get_price_alerts: 'price alerts', create_price_alert: 'create price alert', update_price_alert: 'update price alert', delete_price_alert: 'delete price alert', portfolio_diagnosis: 'portfolio diagnosis', tool_search: 'available tools' }, events: { contextCompressed: 'Context compressed and prepared', contextPrepared: 'Context prepared', step: 'Step {{step}}', toolStart: 'Retrieving: {{name}}', toolDone: 'Data ready: {{name}}', toolFailed: 'Retrieval failed: {{name}}', modelUsage: 'Model usage: input {{input}}, output {{output}}{{extra}}', cache: 'Cached {{count}}', reasoning: 'Reasoning {{count}}', approval: 'Waiting for approval', paused: 'Task paused', done: 'Result ready', error: 'Task failed', started: 'Task started', title: 'Research progress' }, developer: { title: 'Developer details' }, status: { done: 'Completed', error: 'Failed', paused: 'Waiting to resume', running: 'Running' }, toolCalls: '{{count}} tool calls' }, }, }, p5: { @@ -340,6 +340,10 @@ export const configuration = { }, assistantPage: { tools: { get_portfolio: 'Querying portfolio…', get_stock_quote: 'Querying quote…', get_kline_summary: 'Analyzing candlesticks…', get_stock_news: 'Searching related news…', create_price_alert: 'Creating price alert…', get_technical_analysis: 'Analyzing technicals…', get_stock_suggestions: 'Querying AI suggestions…', get_watchlist: 'Querying watchlist…' }, + result: { + facts: 'Evidence-backed facts', inferences: 'Analysis', risks: 'Risks and gaps', evidence: '{{count}} evidence items', nextActions: 'Next steps', observedAt: 'Retrieved {{time}}', dataAt: 'Data as of {{time}}', coverage: 'Covers {{start}} to {{end}}', configureAlert: 'Configure alert', targetPrice: 'Target price', above: 'at or above', below: 'at or below', requestApproval: 'Request approval', alertPrompt: 'Create an intraday alert for {{target}} when the price is {{direction}} {{price}}', + freshness: { fresh: 'Fresh', delayed: 'Delayed', stale: 'May be stale', unknown: 'Time unknown' }, + }, errors: { ai_quota_exhausted: 'The AI service quota is exhausted. Add credits or switch to an available model.', ai_rate_limited: 'The AI service is receiving too many requests. Wait and retry or switch models.', diff --git a/frontend/src/i18n/locales/zh-CN/configuration.ts b/frontend/src/i18n/locales/zh-CN/configuration.ts index db8e54bf..fddf17f5 100644 --- a/frontend/src/i18n/locales/zh-CN/configuration.ts +++ b/frontend/src/i18n/locales/zh-CN/configuration.ts @@ -320,7 +320,7 @@ export const configuration = { permissionsDrawer: { loadFailed: '无法加载工具权限，请稍后重试。', saveFailed: '保存权限设置失败，请重试。', title: '小助手配置', description: '管理工具权限，以及上下文压缩使用的模型和预算。', loading: '正在加载工具权限…' }, contextPanel: { sections: { system: '系统指令', summary: '结构化摘要', page_context: '页面上下文', tool_definitions: '工具定义', history: '历史消息', recent_messages: '最近消息' }, statuses: { not_needed: '本次无需压缩', compressed: '压缩完成', no_gain: '本次压缩无收益，已保留原上下文' }, modes: { balanced: '平衡压缩', preserve_details: '保留细节并压缩', handoff: '整理为交接摘要' }, measurement: { provider: '实际输入 Token', tokenizer: 'Tokenizer 输入 Token', estimate: '估算输入 Token' }, title: '上下文用量', description: '只影响下一次运行，不会删除对话消息。', close: '关闭上下文面板', measuring: '正在测量上下文…', needsCompression: '需要压缩', warning: '接近上限', normal: '正常', snapshot: '摘要 v{{version}}', goal: '目标：{{value}}', state: '状态：{{value}}', todo: '待办：{{value}}', savedTokens: '{{before}} → {{after}}，节省 {{saved}} Token（{{percent}}%）', compressing: '压缩中…', empty: '当前还没有可测量的会话。' }, contextUsage: { states: { normal: '上下文正常', warning: '上下文偏大', needs_compression: '需要压缩' }, aria: '查看上下文用量', title: '{{state}}，估算输入 Token {{used}} / {{budget}}' }, - trace: { extension: { started: '研究可用工具', exposure: '工具目录已准备：{{direct}} 个直达，{{loaded}} 个已加载', candidates: '筛选工具候选：{{count}} 个', completed: '工具研究完成：选出 {{count}} 个', searched: '工具搜索完成：加载 {{count}} 个', fallback: '工具研究回退，继续使用默认工具集', unknown: '扩展事件：{{event}}' }, events: { contextCompressed: '上下文已压缩并准备', contextPrepared: '上下文已准备', step: '执行步骤 {{step}}', toolStart: '调用工具：{{name}}', toolDone: '工具完成：{{name}}', toolFailed: '工具失败：{{name}}', modelUsage: '模型用量：输入 {{input}}，输出 {{output}}{{extra}}', cache: '缓存 {{count}}', reasoning: '推理 {{count}}', approval: '等待用户审批', paused: '任务已暂停', done: '任务完成', error: '任务失败', started: '任务已启动', title: '执行记录' }, status: { done: '已完成', error: '已失败', paused: '等待继续', running: '执行中' }, toolCalls: '{{count}} 次工具调用' }, + trace: { extension: { started: '研究可用工具', exposure: '工具目录已准备：{{direct}} 个直达，{{loaded}} 个已加载', candidates: '筛选工具候选：{{count}} 个', completed: '工具研究完成：选出 {{count}} 个', searched: '工具搜索完成：加载 {{count}} 个', fallback: '工具研究回退，继续使用默认工具集', unknown: '扩展事件：{{event}}' }, tools: { get_portfolio: '持仓', get_stock_quote: '实时行情', find_research_candidates: '研究候选', get_kline_summary: 'K 线走势', get_stock_news: '近期新闻', search_stocks: '股票搜索', get_market_status: '市场状态', get_hot_stocks: '热门股票', get_hot_boards: '热门板块', get_board_stocks: '板块成分股', get_stock_fundamentals: '基本面', get_capital_flow: '资金流向', get_dragon_tiger: '龙虎榜', get_price_alerts: '价格提醒', create_price_alert: '创建价格提醒', update_price_alert: '修改价格提醒', delete_price_alert: '删除价格提醒', portfolio_diagnosis: '持仓诊断', tool_search: '可用工具' }, events: { contextCompressed: '上下文已压缩并准备', contextPrepared: '上下文已准备', step: '执行步骤 {{step}}', toolStart: '正在查询：{{name}}', toolDone: '数据已就绪：{{name}}', toolFailed: '查询失败：{{name}}', modelUsage: '模型用量：输入 {{input}}，输出 {{output}}{{extra}}', cache: '缓存 {{count}}', reasoning: '推理 {{count}}', approval: '等待用户审批', paused: '任务已暂停', done: '结果已生成', error: '任务失败', started: '任务已启动', title: '研究进度' }, developer: { title: '开发者详情' }, status: { done: '已完成', error: '已失败', paused: '等待继续', running: '执行中' }, toolCalls: '{{count}} 次工具调用' }, }, }, p5: { @@ -337,6 +337,10 @@ export const configuration = { }, assistantPage: { tools: { get_portfolio: '正在查询持仓…', get_stock_quote: '正在查询行情…', get_kline_summary: '正在分析 K 线…', get_stock_news: '正在检索相关新闻…', create_price_alert: '正在创建价格提醒…', get_technical_analysis: '正在分析技术面…', get_stock_suggestions: '正在查询 AI 建议…', get_watchlist: '正在查询自选股…' }, + result: { + facts: '事实依据', inferences: '分析判断', risks: '风险与不足', evidence: '{{count}} 条数据依据', nextActions: '下一步', observedAt: '获取于 {{time}}', dataAt: '数据截至 {{time}}', coverage: '覆盖 {{start}} 至 {{end}}', configureAlert: '设置提醒条件', targetPrice: '目标价格', above: '达到或高于', below: '跌至或低于', requestApproval: '提交审批', alertPrompt: '为 {{target}} 创建价格{{direction}} {{price}} 时的盘中提醒', + freshness: { fresh: '较新', delayed: '有延迟', stale: '可能过期', unknown: '时间未知' }, + }, errors: { ai_quota_exhausted: 'AI 服务额度已用尽，请充值或切换可用模型后重试。', ai_rate_limited: 'AI 服务请求过于频繁，请稍后重试或切换模型。', diff --git a/frontend/src/lib/assistant-navigation.ts b/frontend/src/lib/assistant-navigation.ts new file mode 100644 index 00000000..3be14abb --- /dev/null +++ b/frontend/src/lib/assistant-navigation.ts @@ -0,0 +1,20 @@ +export interface AssistantPortfolioTarget { + view: 'kline' + symbol: string + market: 'CN' | 'HK' | 'US' +} + +const SYMBOL_PATTERN = /^[A-Za-z0-9.]{1,32}$/ +const MARKETS = new Set(['CN', 'HK', 'US']) + +export function parseAssistantPortfolioTarget( + params: URLSearchParams, +): AssistantPortfolioTarget | null { + if (params.get('view') !== 'kline') return null + const symbol = (params.get('symbol') || '').trim().toUpperCase() + const market = (params.get('market') || '').trim().toUpperCase() + if (!SYMBOL_PATTERN.test(symbol) || !MARKETS.has(market as AssistantPortfolioTarget['market'])) { + return null + } + return { view: 'kline', symbol, market: market as AssistantPortfolioTarget['market'] } +} diff --git a/frontend/src/pages/Assistant.tsx b/frontend/src/pages/Assistant.tsx index f526bad5..daeb2fb9 100644 --- a/frontend/src/pages/Assistant.tsx +++ b/frontend/src/pages/Assistant.tsx @@ -44,6 +44,7 @@ export default function AssistantPage() { conversationIdFromUrl={conversationId} onConversationChange={setConversationId} initialStockContext={launchContext} + onNavigate={(path) => navigate(path)} /> ) } diff --git a/frontend/src/pages/Stocks.tsx b/frontend/src/pages/Stocks.tsx index 441f1f70..9a6ebfba 100644 --- a/frontend/src/pages/Stocks.tsx +++ b/frontend/src/pages/Stocks.tsx @@ -1,4 +1,5 @@ import { useState, useEffect, useRef, useCallback, useMemo } from 'react' +import { useSearchParams } from 'react-router-dom' import { Plus, Trash2, Pencil, Search, X, TrendingUp, Bot, Play, RefreshCw, Wallet, PiggyBank, ArrowUpRight, ArrowDownRight, Building2, ChevronDown, ChevronRight, Cpu, Bell, Clock, Newspaper, ExternalLink, BarChart3, Brain } from 'lucide-react' import { fetchAPI, stocksApi, type AIService, type NotifyChannel } from '@panwatch/api' import { klinesApi } from '@panwatch/api/klines' @@ -28,6 +29,7 @@ import { useTranslation } from 'react-i18next' import { localizeAgentDescription, localizeAgentName } from '@/i18n/agent-labels' import { getCurrentLocale } from '@/i18n' import { marketSignTextClass } from '@/lib/market-colors' +import { parseAssistantPortfolioTarget } from '@/lib/assistant-navigation' interface AgentResult { success?: boolean @@ -382,6 +384,7 @@ const mergePortfolioQuotes = ( } export default function StocksPage() { + const [searchParams, setSearchParams] = useSearchParams() const { t } = useTranslation('configuration') const stockT = t as unknown as (key: string, options?: Record) => string const klineT = (key: string, options?: Record) => @@ -871,6 +874,24 @@ export default function StocksPage() { setKlineDialogOpen(true) }, [klineSummaries]) + const handledAssistantTargetRef = useRef(null) + useEffect(() => { + const target = parseAssistantPortfolioTarget(searchParams) + if (!target) { + handledAssistantTargetRef.current = null + return + } + const key = `${target.market}:${target.symbol}` + if (handledAssistantTargetRef.current === key) return + handledAssistantTargetRef.current = key + openKlineDialog(target.symbol, target.market) + const next = new URLSearchParams(searchParams) + next.delete('view') + next.delete('symbol') + next.delete('market') + setSearchParams(next, { replace: true }) + }, [openKlineDialog, searchParams, setSearchParams]) + // Open news dialog - pass stock name for more stable search const openNewsDialog = useCallback((stockName?: string) => { setNewsDialogSymbol(stockName || '') // 存储名称用于 UI 显示 diff --git a/frontend/tests/api/chatStream.test.ts b/frontend/tests/api/chatStream.test.ts index f7f72e92..509dc522 100644 --- a/frontend/tests/api/chatStream.test.ts +++ b/frontend/tests/api/chatStream.test.ts @@ -50,6 +50,7 @@ describe('assistant task stream', () => { message_id: 7, content: '完成', created_at: '', + result: null, }) }) @@ -90,6 +91,38 @@ describe('assistant task stream', () => { message_id: 10, content: '全部完成', created_at: '', + result: null, + }) + }) + + it('preserves a structured result on the terminal event', async () => { + const result = { + schema_version: 1, + summary: '结论', + facts: [], + inferences: [], + risks: [], + missing_data: [], + evidence: [], + next_actions: [], + } + readSSE.mockImplementation(async (_path: string, options: { onEvent: (event: unknown) => void }) => { + options.onEvent({ + id: 4, + event: 'done', + data: { message_id: 7, content: '完成', created_at: '', result }, + }) + return { lastEventId: 4 } + }) + + const onDone = vi.fn() + await chatApi.sendAssistantMessageStream(1, '分析市场', { onDone }) + + expect(onDone).toHaveBeenCalledWith({ + message_id: 7, + content: '完成', + created_at: '', + result, }) }) diff --git a/frontend/tests/assistant/AssistantResultCard.test.tsx b/frontend/tests/assistant/AssistantResultCard.test.tsx new file mode 100644 index 00000000..26183035 --- /dev/null +++ b/frontend/tests/assistant/AssistantResultCard.test.tsx @@ -0,0 +1,79 @@ +import { render, screen } from '@testing-library/react' +import userEvent from '@testing-library/user-event' +import { describe, expect, it, vi } from 'vitest' + +import { AssistantResultCard } from '@/components/assistant/AssistantResultCard' +import type { AssistantResult } from '@panwatch/api' + +const result: AssistantResult = { + schema_version: 1, + summary: '贵州茅台行情结论', + facts: [{ text: 'CN:600519 最新价为 100。', evidence_ids: ['ev-1'] }], + inferences: ['短期趋势偏强。'], + risks: ['行情存在波动。'], + missing_data: [], + evidence: [{ + id: 'ev-1', + tool_name: 'get_stock_quote', + source_name: 'PanWatch 行情', + source_url: 'https://example.com/quote', + summary: '最新价 100', + observed_at: '2026-09-29T01:00:00Z', + data_at: '2026-09-29T01:00:00Z', + freshness: 'fresh', + freshness_basis: 'as_of', + symbol: '600519', + market: 'CN', + }], + next_actions: [{ + id: 'open-chart', + kind: 'navigate', + label: '打开 K 线', + payload: { path: '/portfolio?view=kline&symbol=600519&market=CN' }, + requires_approval: false, + }], +} + +describe('AssistantResultCard', () => { + it('shows evidence and accepts a validated chart deep link', async () => { + const user = userEvent.setup() + const onNavigate = vi.fn() + render( + , + ) + + expect(screen.getByText('CN:600519 最新价为 100。')).toBeTruthy() + await user.click(screen.getByRole('button', { name: /依据/ })) + expect(screen.getByRole('link', { name: /PanWatch 行情/ }).getAttribute('href')).toBe('https://example.com/quote') + + await user.click(screen.getByRole('button', { name: '打开 K 线' })) + expect(onNavigate).toHaveBeenCalledWith('/portfolio?view=kline&symbol=600519&market=CN') + }) + + it('rejects an external navigation action', async () => { + const user = userEvent.setup() + const onNavigate = vi.fn() + render( + , + ) + + await user.click(screen.getByRole('button', { name: '打开 K 线' })) + expect(onNavigate).not.toHaveBeenCalled() + }) +}) diff --git a/frontend/tests/assistant/ChatWidgetLayout.test.tsx b/frontend/tests/assistant/ChatWidgetLayout.test.tsx index c6b55df0..c172a61f 100644 --- a/frontend/tests/assistant/ChatWidgetLayout.test.tsx +++ b/frontend/tests/assistant/ChatWidgetLayout.test.tsx @@ -234,10 +234,10 @@ describe('ChatWidget layout', () => { await screen.findByText('已完成分析') expect(screen.getAllByTestId('assistant-trace')).toHaveLength(1) - expect(screen.queryByText('调用工具：get_portfolio')).toBeNull() + expect(screen.queryByText('正在查询：持仓')).toBeNull() - await user.click(screen.getByRole('button', { name: /执行记录/ })) - expect(screen.getByText('调用工具：get_portfolio')).toBeTruthy() + await user.click(screen.getByRole('button', { name: /研究进度/ })) + expect(screen.getByText('正在查询：持仓')).toBeTruthy() }) it('does not render a generic retry card when a stream fails', async () => { diff --git a/frontend/tests/assistant/TraceTimeline.test.tsx b/frontend/tests/assistant/TraceTimeline.test.tsx index 41b7019b..2f693208 100644 --- a/frontend/tests/assistant/TraceTimeline.test.tsx +++ b/frontend/tests/assistant/TraceTimeline.test.tsx @@ -37,7 +37,8 @@ describe('TraceTimeline', () => { />, ) - await user.click(screen.getByRole('button', { name: /执行记录/ })) + await user.click(screen.getByRole('button', { name: /研究进度/ })) + await user.click(screen.getByRole('button', { name: /开发者详情/ })) expect(screen.getByText('模型用量：输入 120，输出 30')).toBeTruthy() }) @@ -54,10 +55,11 @@ describe('TraceTimeline', () => { />, ) - expect(screen.getByText('已完成 · 1.2s · 1 次工具调用 · 150 tokens')).toBeTruthy() - await user.click(screen.getByRole('button', { name: /执行记录/ })) + expect(screen.getByText('已完成 · 1.2s')).toBeTruthy() + await user.click(screen.getByRole('button', { name: /研究进度/ })) + await user.click(screen.getByRole('button', { name: /开发者详情/ })) expect(screen.getByText('模型用量：输入 120，输出 30 · 800ms')).toBeTruthy() - expect(screen.getByText('工具完成：get_portfolio · 420ms')).toBeTruthy() + expect(screen.getAllByText('数据已就绪：持仓 · 420ms').length).toBeGreaterThan(0) }) it('expands the factual steps from the compact summary', async () => { @@ -80,9 +82,10 @@ describe('TraceTimeline', () => { />, ) - await user.click(screen.getByRole('button', { name: /执行记录/ })) + await user.click(screen.getByRole('button', { name: /研究进度/ })) + await user.click(screen.getByRole('button', { name: /开发者详情/ })) - expect(screen.getByText('调用工具：get_portfolio')).toBeTruthy() + expect(screen.getAllByText('正在查询：持仓').length).toBeGreaterThan(0) expect(screen.getByText('{"market":"CN"}')).toBeTruthy() expect(screen.getByText('持仓查询完成')).toBeTruthy() expect(screen.getByText('工具研究完成：选出 1 个')).toBeTruthy() @@ -115,10 +118,11 @@ describe('TraceTimeline', () => { />, ) - await user.click(screen.getByRole('button', { name: /执行记录/ })) + await user.click(screen.getByRole('button', { name: /研究进度/ })) + await user.click(screen.getByRole('button', { name: /开发者详情/ })) expect(screen.getByText('工具目录已准备：1 个直达，0 个已加载')).toBeTruthy() expect(screen.getByText('工具搜索完成：加载 1 个')).toBeTruthy() - expect(screen.getByText('调用工具：get_fundamentals')).toBeTruthy() + expect(screen.getAllByText('正在查询：get_fundamentals').length).toBeGreaterThan(0) }) }) diff --git a/frontend/tests/assistant/assistant-navigation.test.ts b/frontend/tests/assistant/assistant-navigation.test.ts new file mode 100644 index 00000000..e1ea76db --- /dev/null +++ b/frontend/tests/assistant/assistant-navigation.test.ts @@ -0,0 +1,19 @@ +import { describe, expect, it } from 'vitest' + +import { parseAssistantPortfolioTarget } from '@/lib/assistant-navigation' + +describe('parseAssistantPortfolioTarget', () => { + it('parses a validated chart target', () => { + expect(parseAssistantPortfolioTarget(new URLSearchParams('view=kline&symbol=brk.b&market=us'))).toEqual({ + view: 'kline', + symbol: 'BRK.B', + market: 'US', + }) + }) + + it('rejects unsupported views, markets, and symbols', () => { + expect(parseAssistantPortfolioTarget(new URLSearchParams('view=analysis&symbol=600519&market=CN'))).toBeNull() + expect(parseAssistantPortfolioTarget(new URLSearchParams('view=kline&symbol=600519&market=XX'))).toBeNull() + expect(parseAssistantPortfolioTarget(new URLSearchParams('view=kline&symbol=../bad&market=CN'))).toBeNull() + }) +}) diff --git a/packages/pan-agent-runtime/src/pan_agent/contracts.py b/packages/pan-agent-runtime/src/pan_agent/contracts.py index c7f843d4..f284295a 100644 --- a/packages/pan-agent-runtime/src/pan_agent/contracts.py +++ b/packages/pan-agent-runtime/src/pan_agent/contracts.py @@ -88,6 +88,10 @@ class EventType(StrEnum): class Source(BaseModel): name: str = Field(min_length=1) url: str | None = None + as_of: str | None = Field(default=None, max_length=200) + published_at: str | None = Field(default=None, max_length=200) + period_start: str | None = Field(default=None, max_length=200) + period_end: str | None = Field(default=None, max_length=200) class ToolSpec(BaseModel): diff --git a/packages/pan-agent-runtime/src/pan_agent/runtime.py b/packages/pan-agent-runtime/src/pan_agent/runtime.py index 91db4627..9910e470 100644 --- a/packages/pan-agent-runtime/src/pan_agent/runtime.py +++ b/packages/pan-agent-runtime/src/pan_agent/runtime.py @@ -239,6 +239,15 @@ async def emit_model_token(token: str) -> None: {"step": current_step, "status": "running"}, ) current_tool_choice = self._tool_choice_for_turn(request, messages) + if current_tool_choice == _REQUIRED_TOOL_CHOICE and not model_tools: + return await self._finish( + sink, + request, + RunStatus.PARTIAL, + answer, + tool_calls, + "permission_denied", + ) model_started_at = time.monotonic() try: turn = await self._run_model_turn( @@ -512,7 +521,20 @@ async def _resolve_model_tools( deadline: float, ) -> tuple[list[ToolSpec], dict[str, tuple[ToolSpec, RuntimeExtension]]]: """Resolve registered tools plus virtual tools owned by extensions.""" - model_tools = self._tools.model_tools(request, self._policy) + raw_allowed_tool_names = request.context.get("allowed_tool_names") + tools_are_restricted = raw_allowed_tool_names is not None + allowed_tool_names = ( + [name for name in raw_allowed_tool_names if isinstance(name, str)] + if isinstance(raw_allowed_tool_names, (list, tuple, set)) + else [] + ) + allowed_tool_name_set = set(allowed_tool_names) + model_tools = self._tools.model_tools( + request, + self._policy, + names=allowed_tool_names if tools_are_restricted else None, + include_deferred=tools_are_restricted, + ) extension_tools: dict[str, tuple[ToolSpec, RuntimeExtension]] = {} for extension in self._extensions: extension_name = getattr(extension, "name", extension.__class__.__name__) @@ -553,14 +575,21 @@ async def emit_extension_event( ) continue if decision is not None and decision.tool_names is not None: + selected_names = list(decision.tool_names) + if tools_are_restricted: + selected_names = [ + name for name in selected_names if name in allowed_tool_name_set + ] model_tools = self._tools.model_tools( request, self._policy, - names=list(decision.tool_names), + names=selected_names, include_deferred=True, ) if decision is not None and decision.additional_tools: for tool in decision.additional_tools: + if tools_are_restricted and tool.name not in allowed_tool_name_set: + continue if tool.name in extension_tools or any( item.name == tool.name for item in model_tools ): @@ -756,6 +785,11 @@ async def _publish_tool_completed( "tool": call.name, "ok": result.ok, "summary": result.summary, + "data": result.model_dump(mode="json")["data"], + "sources": [source.model_dump(mode="json") for source in result.sources], + "observed_at": ( + result.observed_at.isoformat() if result.observed_at else None + ), "error_code": result.error_code, "duration_ms": duration_ms, "attempt_count": attempt_count, diff --git a/packages/pan-agent-runtime/tests/test_runtime.py b/packages/pan-agent-runtime/tests/test_runtime.py index 12b91c66..2a353406 100644 --- a/packages/pan-agent-runtime/tests/test_runtime.py +++ b/packages/pan-agent-runtime/tests/test_runtime.py @@ -60,6 +60,14 @@ async def decide(self, _request, _tool, _call): return ToolPermissionDecision.deny("not allowed") +class AllowPolicy: + def is_tool_visible(self, _request, _tool): + return True + + async def decide(self, _request, _tool, _call): + return ToolPermissionDecision.allow() + + class CapturingModel: def __init__(self): self.received_tools = [] @@ -421,7 +429,11 @@ async def write_note(_request, _arguments): registry = write_registry(write_note) sink = CollectingSink() - result = asyncio.run(AgentRuntime(model, registry).run(request_with_required_tool, sink)) + result = asyncio.run( + AgentRuntime(model, registry, policy=AllowPolicy()).run( + request_with_required_tool, sink + ) + ) assert result.status is RunStatus.COMPLETED assert result.answer == "提醒已成功更新。" @@ -448,7 +460,9 @@ async def run_turn(self, _messages, _tools, _emit_token, tool_choice=None): ) result = asyncio.run( - AgentRuntime(TextOnlyModel(), write_registry(lambda *_: None)).run( + AgentRuntime( + TextOnlyModel(), write_registry(lambda *_: None), policy=AllowPolicy() + ).run( request_with_required_tool, CollectingSink() ) ) diff --git a/src/modules/assistant/api.py b/src/modules/assistant/api.py index 6d1bb05a..d4cb8c3e 100644 --- a/src/modules/assistant/api.py +++ b/src/modules/assistant/api.py @@ -182,6 +182,8 @@ async def publish(self, event) -> None: "name": data.get("tool", ""), "ok": data.get("ok", False), "preview": data.get("summary", ""), + "sources": data.get("sources") or [], + "observed_at": data.get("observed_at"), }, ) ) diff --git a/src/modules/assistant/event_stream.py b/src/modules/assistant/event_stream.py index 7d2c08bd..29f2d77d 100644 --- a/src/modules/assistant/event_stream.py +++ b/src/modules/assistant/event_stream.py @@ -13,27 +13,7 @@ from src.platform.tasking.contracts import TaskEventType, TaskStatus from .repository import AssistantRepository - -_SSE_EVENT_NAMES = { - TaskEventType.TASK_CREATED: "task_created", - TaskEventType.TASK_QUEUED: "task_queued", - TaskEventType.TASK_STARTED: "run_started", - TaskEventType.CONTEXT_PREPARED: "context_prepared", - TaskEventType.STEP_STARTED: "step_started", - TaskEventType.STEP_PROGRESS: "step_updated", - TaskEventType.EXTENSION_EVENT: "extension_event", - TaskEventType.ANSWER_TOKEN: "token", - TaskEventType.MODEL_USAGE: "model_usage", - TaskEventType.TOOL_STARTED: "tool_call_start", - TaskEventType.TOOL_COMPLETED: "tool_result", - TaskEventType.CHECKPOINT_SAVED: "checkpoint_saved", - TaskEventType.APPROVAL_REQUIRED: "approval_required", - TaskEventType.TASK_PAUSED: "paused", - TaskEventType.TASK_RETRY_SCHEDULED: "retry_scheduled", - TaskEventType.TASK_COMPLETED: "done", - TaskEventType.TASK_FAILED: "error", - TaskEventType.TASK_CANCELLED: "cancelled", -} +from .trace import SSE_EVENT_NAMES def _load_task_events( @@ -69,7 +49,7 @@ async def subscribe_task_events( if events: for event in events: cursor = event.sequence - event_type = _SSE_EVENT_NAMES.get(TaskEventType(event.event_type), event.event_type) + event_type = SSE_EVENT_NAMES.get(TaskEventType(event.event_type), event.event_type) yield format_sse_event(event.sequence, event_type, event.data or {}) last_activity = time.monotonic() continue diff --git a/src/modules/assistant/prompt.py b/src/modules/assistant/prompt.py index b8ef318f..46ba47be 100644 --- a/src/modules/assistant/prompt.py +++ b/src/modules/assistant/prompt.py @@ -16,6 +16,10 @@ - 历史助手文本可能只是计划或错误声明；只有工具执行记录和本轮工具返回结果才能证明操作已完成 - 给出明确的观点和理由，并区分数据事实与分析判断 - 涉及买卖建议时说明风险 +- 研究型回答优先使用“结论、数据事实、分析判断、风险与不足”四个简短段落；没有内容的段落可以省略 +- 数据事实只写工具已经返回的内容，分析判断不得写成已经确认的外部事实 +- 数据时间未知、来源缺失或工具失败时，在“风险与不足”中明确说明 +- 不同来源或不同时间点的数据存在冲突时，列出冲突，不要自行选择一个结果冒充确定事实 - 用中文回答，保持简洁，避免冗余 """ diff --git a/src/modules/assistant/repository.py b/src/modules/assistant/repository.py index cf2aaf68..f37885ad 100644 --- a/src/modules/assistant/repository.py +++ b/src/modules/assistant/repository.py @@ -39,6 +39,8 @@ ) from src.platform.tasking.contracts import TaskEvent, TaskEventType, TaskStatus +from .trace import historical_trace_event + class AssistantRepository: def __init__(self, session: Session) -> None: @@ -79,6 +81,86 @@ def list_messages(self, conversation_id: int) -> list[ChatMessage]: .all() ) + def get_message(self, message_id: int) -> ChatMessage | None: + return self._session.query(ChatMessage).filter(ChatMessage.id == message_id).first() + + def message_results(self, conversation_id: int) -> dict[int, dict]: + """Return structured results keyed by their final assistant message.""" + rows = ( + self._session.query(AssistantTaskRun) + .filter( + AssistantTaskRun.conversation_id == conversation_id, + AssistantTaskRun.final_message_id.isnot(None), + AssistantTaskRun.result_data.isnot(None), + ) + .all() + ) + return { + int(row.final_message_id): row.result_data + for row in rows + if row.final_message_id is not None and isinstance(row.result_data, dict) + } + + def message_task_runs(self, conversation_id: int) -> dict[int, AssistantTaskRun]: + """Return completed task runs keyed by their final assistant message.""" + rows = ( + self._session.query(AssistantTaskRun) + .filter( + AssistantTaskRun.conversation_id == conversation_id, + AssistantTaskRun.final_message_id.isnot(None), + ) + .order_by(AssistantTaskRun.created_at.asc()) + .all() + ) + return { + int(row.final_message_id): row + for row in rows + if row.final_message_id is not None + } + + def tool_invocations_for_tasks( + self, task_run_ids: list[int] + ) -> dict[int, list[AssistantToolInvocation]]: + if not task_run_ids: + return {} + rows = ( + self._session.query(AssistantToolInvocation) + .filter(AssistantToolInvocation.task_run_id.in_(task_run_ids)) + .order_by( + AssistantToolInvocation.task_run_id.asc(), + AssistantToolInvocation.created_at.asc(), + ) + .all() + ) + grouped: dict[int, list[AssistantToolInvocation]] = {} + for row in rows: + grouped.setdefault(int(row.task_run_id), []).append(row) + return grouped + + def message_traces(self, conversation_id: int) -> dict[int, list[dict]]: + """Restore the user-visible execution trace for persisted answers.""" + tasks = self.message_task_runs(conversation_id) + if not tasks: + return {} + message_by_task = { + int(task.id): message_id for message_id, task in tasks.items() + } + rows = ( + self._session.query(AssistantTaskEvent) + .filter(AssistantTaskEvent.task_run_id.in_(list(message_by_task))) + .order_by( + AssistantTaskEvent.task_run_id.asc(), + AssistantTaskEvent.sequence.asc(), + ) + .all() + ) + traces: dict[int, list[dict]] = {} + for row in rows: + item = historical_trace_event(row.event_type, row.data, row.sequence) + if item is not None: + traces.setdefault(message_by_task[int(row.task_run_id)], []).append(item) + return traces + def get_latest_context_snapshot( self, conversation_id: int ) -> AssistantContextSnapshot | None: @@ -365,6 +447,9 @@ def record_tool_completed( duration_ms: int = 0, attempt_count: int = 1, error_code: str | None = None, + result_data: dict | None = None, + sources: list[dict] | None = None, + observed_at: str | datetime | None = None, ) -> AssistantToolInvocation: invocation = ( self._session.query(AssistantToolInvocation) @@ -387,6 +472,15 @@ def record_tool_completed( invocation.duration_ms = max(0, int(duration_ms or 0)) invocation.attempt_count = max(1, int(attempt_count or 1)) invocation.error_code = error_code + invocation.result_data = result_data or {} + invocation.source_data = sources or [] + if isinstance(observed_at, str) and observed_at: + try: + invocation.observed_at = datetime.fromisoformat(observed_at.replace("Z", "+00:00")) + except ValueError: + invocation.observed_at = None + elif isinstance(observed_at, datetime): + invocation.observed_at = observed_at invocation.completed_at = datetime.now(timezone.utc) self.append_task_event( task_run_id, @@ -402,6 +496,10 @@ def record_tool_completed( "duration_ms": invocation.duration_ms, "attempt_count": invocation.attempt_count, "error_code": error_code, + "sources": invocation.source_data, + "observed_at": ( + invocation.observed_at.isoformat() if invocation.observed_at else None + ), }, commit=False, ) @@ -503,6 +601,14 @@ def list_recent_tool_findings( .all() ) + def list_task_tool_invocations(self, task_run_id: int) -> list[AssistantToolInvocation]: + return ( + self._session.query(AssistantToolInvocation) + .filter(AssistantToolInvocation.task_run_id == task_run_id) + .order_by(AssistantToolInvocation.created_at.asc()) + .all() + ) + def save_checkpoint( self, task_run_id: int, @@ -804,7 +910,12 @@ def finish_task( self._session.commit() def complete_task_with_message( - self, task_run_id: int, conversation_id: int, content: str + self, + task_run_id: int, + conversation_id: int, + content: str, + *, + result_data: dict | None = None, ) -> ChatMessage | None: """Commit the final message and task transition as one cancellation-safe unit.""" task = self._require_task(task_run_id) @@ -813,11 +924,13 @@ def complete_task_with_message( conversation = self.get_conversation(conversation_id) if conversation is None: raise LookupError(f"conversation {conversation_id} not found") - conversation.updated_at = datetime.now(timezone.utc) + completed_at = datetime.now(timezone.utc) + conversation.updated_at = completed_at message = ChatMessage( conversation_id=conversation.id, role="assistant", content=content, + created_at=completed_at, ) self._session.add(message) self._session.flush() @@ -835,7 +948,9 @@ def complete_task_with_message( "error_code": None, "checkpoint": None, "checkpoint_id": "", - "finished_at": datetime.now(timezone.utc), + "finished_at": completed_at, + "result_schema_version": int((result_data or {}).get("schema_version") or 1), + "result_data": result_data, }, synchronize_session=False, ) @@ -853,6 +968,8 @@ def complete_task_with_message( data={ "message_id": message.id, "content": content, + "created_at": completed_at.isoformat(), + "result": result_data, **self._task_metrics(completed_task), }, ) @@ -880,6 +997,7 @@ def get_task_snapshot(self, task_run_id: int) -> dict: "retry_count": int(task.retry_count or 0), "context": task.context or {}, "error_code": task.error_code, + "result": task.result_data, "model": task.model, "duration_ms": self._task_duration_ms(task), "usage": { @@ -916,6 +1034,8 @@ def get_task_snapshot(self, task_run_id: int) -> dict: "duration_ms": int(tool.duration_ms or 0), "attempt_count": int(tool.attempt_count or 1), "error_code": tool.error_code, + "sources": tool.source_data or [], + "observed_at": tool.observed_at, } for tool in tools ], diff --git a/src/modules/assistant/result_builder.py b/src/modules/assistant/result_builder.py new file mode 100644 index 00000000..6e56d89b --- /dev/null +++ b/src/modules/assistant/result_builder.py @@ -0,0 +1,631 @@ +"""Build trusted assistant result metadata from durable tool evidence.""" + +from __future__ import annotations + +import asyncio +import json +import logging +import re +from datetime import datetime, timezone +from typing import Any +from urllib.parse import urlencode + +from pydantic import BaseModel, Field + +from .result_schemas import ( + AssistantEvidence, + AssistantFact, + AssistantNextAction, + AssistantResult, +) +from .tool_descriptors import PANWATCH_TOOL_DESCRIPTORS + +logger = logging.getLogger(__name__) + +_DESCRIPTORS = {item.tool_name: item for item in PANWATCH_TOOL_DESCRIPTORS} +_MARKDOWN_PREFIX = re.compile(r"^\s{0,3}(?:#{1,6}|[-*>])\s*") + + +class _ComposedSections(BaseModel): + summary: str = "" + inferences: list[str] = Field(default_factory=list) + risks: list[str] = Field(default_factory=list) + missing_data: list[str] = Field(default_factory=list) + + +def _aware(value: datetime | None) -> datetime | None: + if value is None: + return None + return value if value.tzinfo else value.replace(tzinfo=timezone.utc) + + +def _parse_datetime(value: object) -> datetime | None: + if isinstance(value, datetime): + return _aware(value) + if not isinstance(value, str) or not value.strip(): + return None + raw = value.strip() + try: + return _aware(datetime.fromisoformat(raw.replace("Z", "+00:00"))) + except ValueError: + pass + if re.fullmatch(r"\d{8}", raw): + try: + return datetime.strptime(raw, "%Y%m%d").replace(tzinfo=timezone.utc) + except ValueError: + return None + quarter = re.fullmatch(r"(\d{4})[- ]?Q([1-4])", raw, re.IGNORECASE) + if quarter: + month = int(quarter.group(2)) * 3 + day = 31 if month in {3, 12} else 30 + return datetime(int(quarter.group(1)), month, day, tzinfo=timezone.utc) + return None + + +def _freshness(tool_name: str, data_time: datetime | None, *, date_only: bool = False) -> str: + data_time = _aware(data_time) + if data_time is None: + return "unknown" + descriptor = _DESCRIPTORS.get(tool_name) + policy = str(getattr(descriptor, "data_freshness", "") or "") + age_seconds = max(0.0, (datetime.now(timezone.utc) - data_time).total_seconds()) + age_days = max(0, (datetime.now(timezone.utc).date() - data_time.date()).days) + if policy == "static": + return "fresh" + if tool_name == "get_stock_fundamentals": + if age_days <= 120: + return "fresh" + if age_days <= 200: + return "delayed" + return "stale" + if tool_name == "get_stock_news": + if age_seconds <= 24 * 60 * 60: + return "fresh" + if age_seconds <= 7 * 24 * 60 * 60: + return "delayed" + return "stale" + if tool_name == "get_kline_summary" and date_only: + if age_days <= 4: + return "fresh" + if age_days <= 10: + return "delayed" + return "stale" + if tool_name in {"get_capital_flow", "get_dragon_tiger"} and date_only: + if age_days <= 1: + return "fresh" + if age_days <= 4: + return "delayed" + return "stale" + if date_only: + if age_days <= 1: + return "fresh" + if age_days <= 3: + return "delayed" + return "stale" + if policy == "real_time": + if age_seconds <= 10 * 60: + return "fresh" + if age_seconds <= 60 * 60: + return "delayed" + return "stale" + if policy == "near_real_time": + if age_seconds <= 2 * 60 * 60: + return "fresh" + if age_seconds <= 48 * 60 * 60: + return "delayed" + return "stale" + return "unknown" + + +def _first_paragraph(answer: str) -> str: + for block in re.split(r"\n\s*\n", answer.strip()): + text = " ".join( + _MARKDOWN_PREFIX.sub("", line).strip() + for line in block.splitlines() + if line.strip() + ).strip() + if text: + return text[:500] + return "" + + +def _parse_json_object(raw: str) -> dict[str, Any]: + text = raw.strip() + fenced = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL) + if fenced: + text = fenced.group(1) + else: + start, end = text.find("{"), text.rfind("}") + if start >= 0 and end > start: + text = text[start : end + 1] + value = json.loads(text) + if not isinstance(value, dict): + raise ValueError("assistant result composer did not return an object") + return value + + +def _evidence_and_facts( + invocations: list[Any], + language: str, +) -> tuple[list[AssistantEvidence], list[AssistantFact], list[str], list[str]]: + evidence: list[AssistantEvidence] = [] + facts: list[AssistantFact] = [] + missing_data: list[str] = [] + deterministic_risks: list[str] = [] + quote_points: dict[str, list[tuple[float, datetime | None]]] = {} + kline_points: dict[str, list[tuple[float, str | None]]] = {} + for invocation in invocations: + summary = str(invocation.summary or "").strip()[:2_000] + if invocation.status != "completed": + if summary: + missing_data.append(summary) + continue + sources = invocation.source_data if isinstance(invocation.source_data, list) else [] + if not sources: + sources = [{"name": invocation.tool_name}] + evidence_ids: list[str] = [] + arguments = invocation.arguments if isinstance(invocation.arguments, dict) else {} + for index, source in enumerate(sources): + source = source if isinstance(source, dict) else {"name": str(source)} + evidence_id = f"ev_{invocation.call_id}_{index + 1}" + evidence_ids.append(evidence_id) + published_at = source.get("published_at") + as_of = source.get("as_of") + observed_at = _aware(invocation.observed_at) + if published_at: + data_at = str(published_at) + freshness_basis = "published_at" + reference_time = _parse_datetime(published_at) + elif as_of: + data_at = str(as_of) + freshness_basis = "as_of" + reference_time = _parse_datetime(as_of) + else: + data_at = None + freshness_basis = "observed_at" if observed_at else "unknown" + reference_time = observed_at + evidence.append( + AssistantEvidence( + id=evidence_id, + tool_name=invocation.tool_name, + source_name=str(source.get("name") or invocation.tool_name), + source_url=str(source.get("url") or "").strip() or None, + summary=summary, + observed_at=observed_at, + data_at=data_at, + period_start=str(source.get("period_start") or "").strip() or None, + period_end=str(source.get("period_end") or "").strip() or None, + freshness=_freshness( + invocation.tool_name, + reference_time, + date_only=bool(data_at and re.fullmatch(r"\d{4}-\d{2}-\d{2}", data_at)), + ), + freshness_basis=freshness_basis, + symbol=str(arguments.get("symbol") or "").strip().upper() or None, + market=str(arguments.get("market") or "").strip().upper() or None, + ) + ) + raw_result_data = getattr(invocation, "result_data", None) + data = raw_result_data if isinstance(raw_result_data, dict) else {} + symbol = str(data.get("symbol") or arguments.get("symbol") or "").strip().upper() + market = str(data.get("market") or arguments.get("market") or "CN").strip().upper() + target = f"{market}:{symbol}" if symbol else "" + tool_name = invocation.tool_name + fact_text = "" + + if not data: + fact_text = summary + if summary: + missing_data.append( + _localized( + language, + "该历史记录未保存字段级工具结果，只能恢复当时的工具摘要。", + "This historical record did not preserve field-level tool data; only the original tool summary can be restored.", + ) + ) + elif tool_name == "get_stock_quote": + price = _number(data.get("current_price")) + change_pct = _number(data.get("change_pct")) + if price is None: + missing_data.append( + _localized(language, f"{target or '该标的'} 缺少最新价。", f"The latest price is missing for {target or 'the symbol'}."), + ) + else: + fact_text = _localized( + language, + f"{target} 最新价为 {_display_number(price)}" + + (f"，涨跌幅为 {_display_number(change_pct)}%。" if change_pct is not None else "。"), + f"{target} last traded at {_display_number(price)}" + + (f", with a {_display_number(change_pct)}% change." if change_pct is not None else "."), + ) + quote_points.setdefault(target, []).append((price, _aware(invocation.observed_at))) + elif tool_name == "get_kline_summary": + close = _number(data.get("last_close")) + as_of = str(data.get("asof") or data.get("date") or "").strip() or None + trend = str(data.get("trend") or "").strip() + parts = [] + if close is not None: + parts.append(_localized(language, f"收盘价 {_display_number(close)}", f"close {_display_number(close)}")) + kline_points.setdefault(target, []).append((close, as_of)) + if trend: + parts.append(_localized(language, f"趋势为{trend}", f"trend: {trend}")) + if parts: + prefix = f"{target} K 线" if target else _localized(language, "K 线", "Chart") + fact_text = f"{prefix}（{as_of}）：" + _localized(language, "，".join(parts) + "。", ", ".join(parts) + ".") if as_of else f"{prefix}：" + _localized(language, "，".join(parts) + "。", ", ".join(parts) + ".") + if close is None: + missing_data.append(_localized(language, f"{target or '该标的'} K 线缺少最新收盘价。", f"The chart data for {target or 'the symbol'} lacks a latest close.")) + if as_of is None: + missing_data.append(_localized(language, f"{target or '该标的'} K 线缺少数据截止日期。", f"The chart data for {target or 'the symbol'} lacks an as-of date.")) + elif tool_name == "get_stock_news": + items = data.get("items") if isinstance(data.get("items"), list) else [] + count = len(items) + fact_text = _localized( + language, + f"{target} 近 7 天检索到 {count} 条相关新闻。", + f"The search found {count} related news items for {target} in the last 7 days.", + ) + if count == 0: + missing_data.append(_localized(language, f"{target} 近 7 天未检索到新闻，事件风险仍需从其他渠道核验。", f"No news was found for {target} in the last 7 days; event risk still needs verification elsewhere.")) + elif tool_name == "get_stock_fundamentals": + period = str(data.get("report_date") or data.get("report_period") or data.get("reporting_period") or "").strip() + labels = (("pe_ratio", "PE"), ("pb_ratio", "PB"), ("roe", "ROE"), ("revenue", _localized(language, "营收", "revenue")), ("net_profit", _localized(language, "净利润", "net profit"))) + values = [f"{label} {_display_number(value)}" for key, label in labels if (value := _number(data.get(key))) is not None] + if period or values: + prefix = _localized(language, f"{target} 基本面", f"{target} fundamentals") + period_text = _localized(language, f"报告期 {period}", f"reporting period {period}") if period else "" + fact_text = f"{prefix}：" + _localized(language, "，".join([item for item in [period_text, *values] if item]) + "。", ", ".join([item for item in [period_text, *values] if item]) + ".") + if not period: + missing_data.append(_localized(language, f"{target} 基本面数据缺少报告期。", f"The fundamentals for {target} lack a reporting period.")) + elif tool_name == "get_capital_flow": + main_net = _number(data.get("main_net_inflow")) + main_pct = _number(data.get("main_net_inflow_pct")) + date = str(data.get("date") or data.get("trade_date") or "").strip() + if main_net is not None: + fact_text = _localized( + language, + f"{target} 主力净流入为 {_display_number(main_net)}" + + (f"（{_display_number(main_pct)}%）" if main_pct is not None else "") + + (f"，数据日期 {date}。" if date else "。"), + f"{target} main net inflow was {_display_number(main_net)}" + + (f" ({_display_number(main_pct)}%)" if main_pct is not None else "") + + (f" on {date}." if date else "."), + ) + else: + missing_data.append(_localized(language, f"{target} 资金流数据缺少主力净流入。", f"Capital flow data for {target} lacks main net inflow.")) + elif tool_name == "find_research_candidates": + count = int(_number(data.get("count")) or 0) + snapshot_date = str(data.get("snapshot_date") or "").strip() + fact_text = _localized( + language, + f"机会筛选得到 {count} 个研究候选" + (f"，快照日期 {snapshot_date}。" if snapshot_date else "。"), + f"Opportunity screening returned {count} research candidates" + (f" for snapshot {snapshot_date}." if snapshot_date else "."), + ) + if not snapshot_date: + missing_data.append(_localized(language, "机会筛选结果缺少快照日期。", "The opportunity screening result lacks a snapshot date.")) + elif summary: + fact_text = summary + + if fact_text: + facts.append(AssistantFact(text=fact_text, evidence_ids=evidence_ids)) + + for target, points in quote_points.items(): + if len(points) < 2: + continue + prices = [item[0] for item in points] + low, high = min(prices), max(prices) + if low > 0 and (high - low) / low >= 0.005: + deterministic_risks.append( + _localized( + language, + f"{target} 的多次行情观测值在 {_display_number(low)}–{_display_number(high)} 之间；它们来自不同观测时点，判断时应使用最新一次。", + f"Repeated quote observations for {target} range from {_display_number(low)} to {_display_number(high)}; they were captured at different times, so use the latest observation.", + ) + ) + for target, quotes in quote_points.items(): + if not quotes or target not in kline_points: + continue + quote_price = quotes[-1][0] + close, as_of = kline_points[target][-1] + if close > 0 and abs(quote_price - close) / close >= 0.03: + deterministic_risks.append( + _localized( + language, + f"{target} 最新价 {_display_number(quote_price)} 与 K 线收盘价 {_display_number(close)}" + + (f"（{as_of}）" if as_of else "") + + "差异较大；两者时点不同，不能直接视为同一价格。", + f"The latest price for {target} ({_display_number(quote_price)}) differs materially from the chart close ({_display_number(close)})" + + (f" on {as_of}" if as_of else "") + + "; they refer to different times and should not be treated as the same price.", + ) + ) + return evidence, facts, missing_data, deterministic_risks + + +def _number(value: object) -> float | None: + if value is None or isinstance(value, bool): + return None + try: + number = float(value) + except (TypeError, ValueError): + return None + return number if number == number and abs(number) != float("inf") else None + + +def _display_number(value: float | None) -> str: + if value is None: + return "—" + return f"{value:,.4f}".rstrip("0").rstrip(".") + + +def _localized(language: str, zh: str, en: str) -> str: + return en if language == "en-US" else zh + + +def _next_actions(task_id: int, invocations: list[Any], language: str) -> list[AssistantNextAction]: + completed_tools = {row.tool_name for row in invocations if row.status == "completed"} + symbol = "" + market = "" + for row in reversed(invocations): + if row.status != "completed": + continue + arguments = row.arguments if isinstance(row.arguments, dict) else {} + if arguments.get("symbol"): + symbol = str(arguments["symbol"]).strip().upper() + market = str(arguments.get("market") or "CN").strip().upper() + break + + actions: list[AssistantNextAction] = [] + + def add(kind: str, label: str, payload: dict, *, approval: bool = False) -> None: + if len(actions) >= 5: + return + actions.append( + AssistantNextAction( + id=f"action_{task_id}_{len(actions) + 1}", + kind=kind, + label=label, + payload=payload, + requires_approval=approval, + ) + ) + + if symbol: + target = f"{market}:{symbol}" + if "get_kline_summary" not in completed_tools: + add( + "follow_up", + _localized(language, "分析 K 线", "Analyze chart"), + { + "prompt": _localized( + language, + f"分析 {target} 的近期 K 线走势和关键支撑压力位", + f"Analyze the recent chart trend and key support and resistance levels for {target}", + ) + }, + ) + if "get_stock_news" not in completed_tools: + add( + "follow_up", + _localized(language, "查看近期消息", "Review recent news"), + { + "prompt": _localized( + language, + f"查询并分析 {target} 最近七天的重要新闻和公告", + f"Find and analyze important news and filings for {target} from the last seven days", + ) + }, + ) + if {"get_kline_summary", "get_stock_news"} <= completed_tools: + add( + "follow_up", + _localized(language, "解释主要风险", "Explain key risks"), + { + "prompt": _localized( + language, + f"基于已有研究，解释 {target} 当前最重要的风险、失效条件和需要继续验证的数据", + f"Based on the existing research, explain the key risks, invalidation conditions, and data that still needs verification for {target}", + ) + }, + ) + add( + "navigate", + _localized(language, "打开 K 线", "Open chart"), + {"path": f"/portfolio?{urlencode({'view': 'kline', 'symbol': symbol, 'market': market})}"}, + ) + add( + "follow_up", + _localized(language, "横向比较", "Compare peers"), + { + "prompt": _localized( + language, + f"将 {target} 与所属板块和主要指数进行对比", + f"Compare {target} with its sector and major market indexes", + ) + }, + ) + add( + "tool_proposal", + _localized(language, "设置价格提醒", "Create price alert"), + { + "tool_name": "create_price_alert", + "arguments": {"symbol": symbol, "market": market}, + "required_inputs": ["direction", "target_price"], + }, + approval=True, + ) + elif "get_portfolio" in completed_tools: + add( + "navigate", + _localized(language, "查看持仓", "Open portfolio"), + {"path": "/portfolio"}, + ) + add( + "follow_up", + _localized(language, "继续诊断风险", "Continue risk review"), + { + "prompt": _localized( + language, + "继续分析持仓集中度、相关性和最大回撤风险", + "Continue analyzing portfolio concentration, correlation, and drawdown risk", + ) + }, + ) + elif "find_research_candidates" in completed_tools: + add( + "navigate", + _localized(language, "查看机会", "Open opportunities"), + {"path": "/opportunities"}, + ) + + if "create_price_alert" in completed_tools or "get_price_alerts" in completed_tools: + actions = [item for item in actions if item.kind != "tool_proposal"] + add("navigate", _localized(language, "查看提醒", "Open alerts"), {"path": "/alerts"}) + return actions[:5] + + +async def _compose_sections( + client: Any, + *, + answer: str, + facts: list[AssistantFact], + missing_data: list[str], +) -> _ComposedSections | None: + if client is None or not facts: + return None + evidence_payload = [fact.model_dump(mode="json") for fact in facts] + system_prompt = ( + "You convert an investment assistant answer into a small JSON result. " + "Use the same language as the answer. Return only a JSON object with keys " + "summary, inferences, risks, missing_data. The supplied evidence facts are immutable " + "and must not be rewritten or expanded. Inferences are analytical conclusions. " + "If supplied facts conflict, describe the conflict in risks instead of silently " + "choosing one. Do not add sources, prices, dates, or claims that are absent from the input." + ) + user_content = json.dumps( + { + "answer": answer[:12_000], + "evidence_facts": evidence_payload, + "known_missing_data": missing_data, + }, + ensure_ascii=False, + ) + try: + raw = await asyncio.wait_for( + client.chat_multi( + [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": user_content}, + ], + temperature=0, + max_tokens=1_200, + ), + timeout=15, + ) + return _ComposedSections.model_validate(_parse_json_object(raw)) + except Exception: + logger.warning( + "Assistant result composition failed; using deterministic result", + exc_info=True, + ) + return None + + +async def build_assistant_result( + *, + task_id: int, + answer: str, + invocations: list[Any], + language: str, + client: Any = None, +) -> AssistantResult: + """Combine trusted evidence with model-organized, validation-bounded sections.""" + deterministic = build_deterministic_assistant_result( + task_id=task_id, + answer=answer, + invocations=invocations, + language=language, + ) + deterministic_facts = deterministic.facts + missing_data = deterministic.missing_data + evidence = deterministic.evidence + composed = await _compose_sections( + client, + answer=answer, + facts=deterministic_facts, + missing_data=missing_data, + ) + combined_missing_data = list(missing_data) + if composed is not None: + combined_missing_data.extend(composed.missing_data) + return AssistantResult( + summary=( + composed.summary.strip()[:1_000] + if composed and composed.summary.strip() + else deterministic.summary + ), + facts=deterministic_facts, + inferences=_clean_unique(list(composed.inferences if composed else []), limit=6), + risks=_clean_unique( + [*deterministic.risks, *(composed.risks if composed else [])], + limit=6, + ), + missing_data=_clean_unique(combined_missing_data, limit=6), + evidence=evidence, + next_actions=deterministic.next_actions, + ) + + +def build_deterministic_assistant_result( + *, + task_id: int, + answer: str, + invocations: list[Any], + language: str, +) -> AssistantResult: + """Build a stable result without an extra model call, including legacy tasks.""" + evidence, deterministic_facts, missing_data, deterministic_risks = _evidence_and_facts( + invocations, + language, + ) + evidence = evidence[:20] + retained_ids = {item.id for item in evidence} + deterministic_facts = [ + AssistantFact( + text=fact.text, + evidence_ids=[item for item in fact.evidence_ids if item in retained_ids], + ) + for fact in deterministic_facts + if any(item in retained_ids for item in fact.evidence_ids) + ] + risks = list(deterministic_risks) + if any(item.freshness in {"stale", "unknown"} for item in evidence): + warning = _localized( + language, + "部分依据缺少可验证的数据时点或已经过期，请结合最新数据复核。", + "Some evidence has an unknown or stale data time; verify it against current data.", + ) + if warning not in risks: + risks.append(warning) + return AssistantResult( + summary=_first_paragraph(answer), + facts=deterministic_facts[:8], + inferences=[], + risks=_clean_unique(risks, limit=6), + missing_data=_clean_unique(missing_data, limit=6), + evidence=evidence, + next_actions=_next_actions(task_id, invocations, language), + ) + + +def _clean_unique(items: list[str], *, limit: int) -> list[str]: + values: list[str] = [] + seen: set[str] = set() + for item in items: + value = item.strip()[:1_000] + if not value or value in seen: + continue + seen.add(value) + values.append(value) + if len(values) >= limit: + break + return values diff --git a/src/modules/assistant/result_schemas.py b/src/modules/assistant/result_schemas.py new file mode 100644 index 00000000..29f110b5 --- /dev/null +++ b/src/modules/assistant/result_schemas.py @@ -0,0 +1,48 @@ +"""Versioned contracts for evidence-backed assistant results.""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any, Literal + +from pydantic import BaseModel, Field + + +class AssistantEvidence(BaseModel): + id: str + tool_name: str + source_name: str + source_url: str | None = None + summary: str + observed_at: datetime | None = None + data_at: str | None = None + period_start: str | None = None + period_end: str | None = None + freshness: Literal["fresh", "delayed", "stale", "unknown"] = "unknown" + freshness_basis: Literal["published_at", "as_of", "observed_at", "unknown"] = "unknown" + symbol: str | None = None + market: str | None = None + + +class AssistantFact(BaseModel): + text: str + evidence_ids: list[str] = Field(default_factory=list) + + +class AssistantNextAction(BaseModel): + id: str + kind: Literal["follow_up", "navigate", "tool_proposal"] + label: str + payload: dict[str, Any] = Field(default_factory=dict) + requires_approval: bool = False + + +class AssistantResult(BaseModel): + schema_version: int = 1 + summary: str = "" + facts: list[AssistantFact] = Field(default_factory=list) + inferences: list[str] = Field(default_factory=list) + risks: list[str] = Field(default_factory=list) + missing_data: list[str] = Field(default_factory=list) + evidence: list[AssistantEvidence] = Field(default_factory=list) + next_actions: list[AssistantNextAction] = Field(default_factory=list) diff --git a/src/modules/assistant/schemas.py b/src/modules/assistant/schemas.py index 417251bb..2ba711bc 100644 --- a/src/modules/assistant/schemas.py +++ b/src/modules/assistant/schemas.py @@ -3,11 +3,13 @@ from __future__ import annotations from datetime import datetime -from typing import Literal +from typing import Any, Literal from pan_agent import ApprovalDecision, PermissionMode, ToolRisk from pydantic import BaseModel, Field +from .result_schemas import AssistantResult + class CreateConversationCommand(BaseModel): stock_symbol: str | None = Field(default=None, max_length=32) @@ -39,6 +41,8 @@ class MessageDTO(BaseModel): role: str content: str created_at: datetime | None = None + result: AssistantResult | None = None + trace: list[dict[str, Any]] | None = None class ConversationDetailDTO(BaseModel): diff --git a/src/modules/assistant/service.py b/src/modules/assistant/service.py index af9eff97..24dc1e8d 100644 --- a/src/modules/assistant/service.py +++ b/src/modules/assistant/service.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re from dataclasses import dataclass from datetime import datetime, timedelta, timezone @@ -52,12 +53,14 @@ from .prompt import build_assistant_messages from .portfolio_diagnosis import PortfolioDiagnosisExtension from .repository import AssistantRepository +from .result_builder import build_deterministic_assistant_result from .schemas import ( ConversationDetailDTO, ConversationDTO, CreateConversationCommand, MessageDTO, ) +from .result_schemas import AssistantResult from .tool_descriptors import localized_tool_descriptors from .tool_adapters import execute_tool from .tools import build_panwatch_tool_registry @@ -167,11 +170,35 @@ def get_suggested_questions(self, symbol: str, market: str = "CN") -> list[str]: def get_conversation(self, conversation_id: int) -> ConversationDetailDTO: conversation = self._require_conversation(conversation_id) + results = self._repository.message_results(conversation_id) + traces = self._repository.message_traces(conversation_id) + tasks = self._repository.message_task_runs(conversation_id) + invocations = self._repository.tool_invocations_for_tasks( + [int(task.id) for task in tasks.values()] + ) + messages = self._repository.list_messages(conversation_id) + language = self._report_language() + for message in messages: + task = tasks.get(int(message.id)) + if message.role != "assistant" or int(message.id) in results or task is None: + continue + task_invocations = invocations.get(int(task.id), []) + if task_invocations: + results[int(message.id)] = build_deterministic_assistant_result( + task_id=int(task.id), + answer=message.content, + invocations=task_invocations, + language=language, + ).model_dump(mode="json") return ConversationDetailDTO( conversation=self._conversation_dto(conversation), messages=[ - self._message_dto(row) - for row in self._repository.list_messages(conversation_id) + self._message_dto( + row, + result=results.get(row.id), + trace=traces.get(row.id), + ) + for row in messages ], ) @@ -669,10 +696,45 @@ def _snapshot_compression_dto(cls, snapshot) -> ContextCompressionDTO | None: def create_task(self, conversation_id: int, user_message_id: int): self._require_conversation(conversation_id) + message = self._repository.get_message(user_message_id) return self._repository.create_task( - conversation_id=conversation_id, user_message_id=user_message_id, context={} + conversation_id=conversation_id, + user_message_id=user_message_id, + context=self._write_action_context(message.content if message else ""), ) + @staticmethod + def _write_action_context(content: str) -> dict: + """Constrain explicit write requests to one approval-protected tool.""" + normalized = content.strip().lower() + alert_request = any(term in normalized for term in ("提醒", "预警", "alert")) + if not alert_request: + return {} + create_request = any( + term in normalized + for term in ("创建", "设置", "新增", "提醒我", "create", "set ") + ) + has_exact_target = re.search( + r"(? None: self._repository.record_tool_completed( task_id, @@ -683,6 +745,9 @@ def record_tool_completion(self, task_id: int, data: dict) -> None: duration_ms=int(data.get("duration_ms") or 0), attempt_count=int(data.get("attempt_count") or 1), error_code=data.get("error_code"), + result_data=data.get("data") or {}, + sources=data.get("sources") or [], + observed_at=data.get("observed_at"), ) def record_tool_started(self, task_id: int, data: dict) -> None: @@ -705,12 +770,27 @@ def record_assistant_message( ) def complete_task_with_message( - self, task_id: int, conversation_id: int, content: str + self, + task_id: int, + conversation_id: int, + content: str, + *, + result: AssistantResult | None = None, ) -> MessageDTO | None: message = self._repository.complete_task_with_message( - task_id, conversation_id, content + task_id, + conversation_id, + content, + result_data=result.model_dump(mode="json") if result else None, + ) + return ( + self._message_dto( + message, + result=result.model_dump(mode="json") if result else None, + ) + if message is not None + else None ) - return self._message_dto(message) if message is not None else None def finish_task(self, task_id: int, result, final_message_id: int) -> None: self._repository.finish_task( @@ -862,10 +942,17 @@ def _conversation_dto(conversation) -> ConversationDTO: ) @staticmethod - def _message_dto(message) -> MessageDTO: + def _message_dto( + message, + *, + result: dict | None = None, + trace: list[dict] | None = None, + ) -> MessageDTO: return MessageDTO( id=message.id, role=message.role, content=message.content, created_at=message.created_at, + result=AssistantResult.model_validate(result) if result else None, + trace=trace, ) diff --git a/src/modules/assistant/task_runner.py b/src/modules/assistant/task_runner.py index 020c9389..9686b526 100644 --- a/src/modules/assistant/task_runner.py +++ b/src/modules/assistant/task_runner.py @@ -25,9 +25,11 @@ from src.platform.ai.errors import descriptor_for_code from src.platform.persistence.database import SessionLocal from src.platform.tasking.contracts import TaskEventType, TaskStatus +from src.platform.language import resolve_report_language from .prompt import build_assistant_messages from .repository import AssistantRepository +from .result_builder import build_assistant_result from .service import AssistantService logger = logging.getLogger(__name__) @@ -44,6 +46,8 @@ "empty_answer": "助手暂时不可用，请稍后重试。", "transport_failed": "助手任务执行失败，请稍后重试。", "worker_cancelled": "助手任务已停止。", + "permission_denied": "当前助手权限不允许执行该操作，请在助手设置中调整后重试。", + "required_tool_call_missing": "未能生成可审批的操作，请重试。", } @@ -217,6 +221,7 @@ async def _run_message(self, task_id: int, conversation_id: int) -> None: return if service._repository.is_task_cancelled(task_id): return + task_run = service._repository.get_task_run(task_id) context_result = await service.prepare_context(conversation_id) if context_result is not None: service._repository.append_task_event( @@ -232,7 +237,8 @@ async def _run_message(self, task_id: int, conversation_id: int) -> None: "compressed_message_count": context_result.compressed_message_count, }, ) - runtime = service.build_runtime(service.build_failover_client()) + client = service.build_failover_client() + runtime = service.build_runtime(client) messages = ( context_result.messages if context_result is not None @@ -246,14 +252,17 @@ async def _run_message(self, task_id: int, conversation_id: int) -> None: request = RunRequest( run_id=str(task_id), messages=messages, - context=( - { - "context_usage": context_result.usage_after.model_dump(mode="json"), - "context_compressed": context_result.compressed, - } - if context_result is not None - else {} - ), + context={ + **dict(task_run.context or {}), + **( + { + "context_usage": context_result.usage_after.model_dump(mode="json"), + "context_compressed": context_result.compressed, + } + if context_result is not None + else {} + ), + }, limits=RunLimits( max_steps=ASSISTANT_MAX_STEPS, max_tool_calls=ASSISTANT_MAX_TOOL_CALLS, @@ -270,7 +279,13 @@ async def _run_message(self, task_id: int, conversation_id: int) -> None: timeout=ASSISTANT_RUN_TIMEOUT_SECONDS, ) await sink.flush() - await self._finish_result(service, task_id, conversation_id, result) + await self._finish_result( + service, + task_id, + conversation_id, + result, + composer_client=client, + ) except TaskCancelledError: await self._flush_sink(sink) self._cancel_if_needed(service, task_id) @@ -300,7 +315,8 @@ async def _run_resume( return if service._repository.is_task_cancelled(task_id): return - runtime = service.build_runtime(service.build_failover_client()) + client = service.build_failover_client() + runtime = service.build_runtime(client) request = RunRequest( run_id=str(task_id), messages=checkpoint.messages, @@ -322,7 +338,13 @@ async def _run_resume( timeout=ASSISTANT_RUN_TIMEOUT_SECONDS, ) await sink.flush() - await self._finish_result(service, task_id, conversation_id, result) + await self._finish_result( + service, + task_id, + conversation_id, + result, + composer_client=client, + ) except TaskCancelledError: await self._flush_sink(sink) self._cancel_if_needed(service, task_id) @@ -342,7 +364,13 @@ async def _run_resume( db.close() async def _finish_result( - self, service: AssistantService, task_id: int, conversation_id: int, result: RunResult + self, + service: AssistantService, + task_id: int, + conversation_id: int, + result: RunResult, + *, + composer_client=None, ) -> None: if result.status is RunStatus.WAITING_FOR_APPROVAL: approvals = service.pause_task(task_id, result) @@ -372,7 +400,36 @@ async def _finish_result( if result.status is not RunStatus.COMPLETED or not result.answer.strip(): self._fail(service, task_id, result.error_code or "empty_answer") return - service.complete_task_with_message(task_id, conversation_id, result.answer) + tokens_before = int(getattr(composer_client, "total_tokens_used", 0) or 0) + structured_result = await build_assistant_result( + task_id=task_id, + answer=result.answer, + invocations=service._repository.list_task_tool_invocations(task_id), + language=resolve_report_language(service._repository.session), + client=composer_client, + ) + tokens_after = int(getattr(composer_client, "total_tokens_used", 0) or 0) + composer_usage = getattr(composer_client, "last_usage", None) + if tokens_after > tokens_before and composer_usage is not None: + usage_data = ( + composer_usage.model_dump(mode="json") + if hasattr(composer_usage, "model_dump") + else dict(composer_usage) + ) + usage_data["phase"] = "result_composition" + service._repository.record_model_usage(task_id, usage_data) + service._repository.append_task_event( + task_id, + TaskEventType.MODEL_USAGE, + status=TaskStatus.RUNNING, + data=usage_data, + ) + service._repository.complete_task_with_message( + task_id, + conversation_id, + result.answer, + result_data=structured_result.model_dump(mode="json"), + ) def _fail(self, service: AssistantService, task_id: int, error_code: str) -> None: descriptor = descriptor_for_code(error_code) diff --git a/src/modules/assistant/tools.py b/src/modules/assistant/tools.py index 0126be5e..7841e822 100644 --- a/src/modules/assistant/tools.py +++ b/src/modules/assistant/tools.py @@ -88,6 +88,16 @@ def _published_at(value: object) -> str: return str(value or "") +def _data_as_of(data: object, *keys: str) -> str | None: + if not isinstance(data, dict): + return None + for key in keys: + value = data.get(key) + if value not in (None, ""): + return _published_at(value)[:200] + return None + + def _json_safe(value: object) -> object: if isinstance(value, datetime): return value.isoformat() @@ -311,14 +321,15 @@ async def get_stock_quote(_request: RunRequest, arguments: dict) -> ToolResult: ) } name = data.get("name") or symbol + observed_at = datetime.now(UTC) return ToolResult.success( summary=( f"{name}（{market.value}:{symbol}）最新价 {data.get('current_price')}，" f"涨跌幅 {data.get('change_pct')}%。" ), data=data, - sources=[{"name": "PanWatch 行情数据"}], - observed_at=datetime.now(UTC), + sources=[{"name": "PanWatch 行情数据", "as_of": observed_at.isoformat()}], + observed_at=observed_at, ) async def get_kline_summary(_request: RunRequest, arguments: dict) -> ToolResult: @@ -342,7 +353,20 @@ async def get_kline_summary(_request: RunRequest, arguments: dict) -> ToolResult return ToolResult.success( summary=f"{market.value}:{symbol} 的 K 线摘要已就绪：{summary}", data=summary, - sources=[{"name": "PanWatch K 线数据"}], + sources=[{ + "name": "PanWatch K 线数据", + "as_of": _data_as_of( + summary, + "asof", + "date", + "trade_date", + "latest_date", + "period_end", + "end_date", + ), + "period_start": _data_as_of(summary, "period_start", "start_date"), + "period_end": _data_as_of(summary, "period_end", "end_date", "asof"), + }], observed_at=datetime.now(UTC), ) @@ -378,7 +402,24 @@ async def get_stock_news(_request: RunRequest, arguments: dict) -> ToolResult: return ToolResult.success( summary=f"{market.value}:{symbol} 近 7 天相关新闻 {len(items)} 条。", data={"symbol": symbol, "market": market.value, "items": items}, - sources=[{"name": "PanWatch 新闻数据"}], + sources=( + [ + { + "name": " · ".join( + value + for value in ( + str(item.get("source") or "").strip(), + str(item.get("title") or "").strip(), + ) + if value + )[:200] or "新闻来源", + "url": item.get("url") or None, + "published_at": str(item.get("published_at") or "")[:200] or None, + } + for item in items + ][:5] + or [{"name": "PanWatch 新闻数据"}] + ), observed_at=datetime.now(UTC), ) @@ -588,7 +629,16 @@ async def get_stock_fundamentals(_request: RunRequest, arguments: dict) -> ToolR return ToolResult.success( summary=f"已获取 {market.value}:{symbol} 的基本面摘要。", data=data, - sources=[{"name": "PanWatch 基本面数据"}], + sources=[{ + "name": "PanWatch 基本面数据", + "as_of": _data_as_of( + data, + "report_date", + "report_period", + "reporting_period", + "date", + ), + }], observed_at=datetime.now(UTC), ) @@ -610,10 +660,14 @@ async def get_capital_flow(_request: RunRequest, arguments: dict) -> ToolResult: summary=f"未找到 {market.value}:{symbol} 的资金流向数据。", error_code="capital_flow_unavailable", ) + data = _json_safe(item) return ToolResult.success( summary=f"已获取 {market.value}:{symbol} 的资金流向摘要。", - data=_json_safe(item), - sources=[{"name": "PanWatch 资金流向"}], + data=data, + sources=[{ + "name": "PanWatch 资金流向", + "as_of": _data_as_of(data, "date", "trade_date", "updated_at"), + }], observed_at=datetime.now(UTC), ) @@ -650,7 +704,7 @@ async def get_dragon_tiger(_request: RunRequest, arguments: dict) -> ToolResult: return ToolResult.success( summary=f"{trade_date} 找到 {len(data)} 条龙虎榜记录。", data={"market": market.value, "date": trade_date, "count": len(data), "items": data}, - sources=[{"name": "PanWatch 龙虎榜"}], + sources=[{"name": "PanWatch 龙虎榜", "as_of": trade_date}], observed_at=datetime.now(UTC), ) diff --git a/src/modules/assistant/trace.py b/src/modules/assistant/trace.py new file mode 100644 index 00000000..fcd2268f --- /dev/null +++ b/src/modules/assistant/trace.py @@ -0,0 +1,62 @@ +"""Public trace projection shared by live SSE and conversation history.""" + +from __future__ import annotations + +from typing import Any + +from src.platform.tasking.contracts import TaskEventType + + +SSE_EVENT_NAMES = { + TaskEventType.TASK_CREATED: "task_created", + TaskEventType.TASK_QUEUED: "task_queued", + TaskEventType.TASK_STARTED: "run_started", + TaskEventType.CONTEXT_PREPARED: "context_prepared", + TaskEventType.STEP_STARTED: "step_started", + TaskEventType.STEP_PROGRESS: "step_updated", + TaskEventType.EXTENSION_EVENT: "extension_event", + TaskEventType.ANSWER_TOKEN: "token", + TaskEventType.MODEL_USAGE: "model_usage", + TaskEventType.TOOL_STARTED: "tool_call_start", + TaskEventType.TOOL_COMPLETED: "tool_result", + TaskEventType.CHECKPOINT_SAVED: "checkpoint_saved", + TaskEventType.APPROVAL_REQUIRED: "approval_required", + TaskEventType.TASK_PAUSED: "paused", + TaskEventType.TASK_RETRY_SCHEDULED: "retry_scheduled", + TaskEventType.TASK_COMPLETED: "done", + TaskEventType.TASK_FAILED: "error", + TaskEventType.TASK_CANCELLED: "cancelled", +} + +_HISTORICAL_TRACE_EVENTS = { + "run_started", + "context_prepared", + "step_updated", + "extension_event", + "model_usage", + "tool_call_start", + "tool_result", + "approval_required", + "paused", + "done", + "error", +} + + +def historical_trace_event( + event_type: str, + data: dict[str, Any] | None, + sequence: int, +) -> dict[str, Any] | None: + """Return the same compact trace item emitted by the browser stream.""" + try: + public_name = SSE_EVENT_NAMES.get(TaskEventType(event_type), event_type) + except ValueError: + public_name = event_type + if public_name not in _HISTORICAL_TRACE_EVENTS: + return None + public_data = dict(data or {}) + if public_name == "done": + public_data.pop("content", None) + public_data.pop("result", None) + return {"event": public_name, "data": public_data, "id": sequence} diff --git a/src/platform/marketdata/collectors/kline_collector.py b/src/platform/marketdata/collectors/kline_collector.py index f16864d0..0b6dfa99 100644 --- a/src/platform/marketdata/collectors/kline_collector.py +++ b/src/platform/marketdata/collectors/kline_collector.py @@ -713,6 +713,8 @@ def get_kline_summary(self, symbol: str) -> dict: "timeframe": "1d", "computed_at": now, "asof": last_date, + "period_start": klines[0].date, + "period_end": last_date, "params": { "ma": [5, 10, 20, 60], "macd": {"fast": 12, "slow": 26, "signal": 9}, diff --git a/src/platform/persistence/migrations.py b/src/platform/persistence/migrations.py index 8492d324..0533a0e2 100644 --- a/src/platform/persistence/migrations.py +++ b/src/platform/persistence/migrations.py @@ -2012,6 +2012,38 @@ def _m127_assistant_trace_metrics(conn: Connection) -> None: _add_column_if_missing(conn, "assistant_tool_invocations", name, statement) +def _m128_assistant_trusted_results(conn: Connection) -> None: + """Persist versioned assistant results and evidence observation times.""" + for name, statement in ( + ( + "result_schema_version", + "ALTER TABLE assistant_task_runs ADD COLUMN result_schema_version INTEGER NOT NULL DEFAULT 1", + ), + ( + "result_data", + "ALTER TABLE assistant_task_runs ADD COLUMN result_data JSON", + ), + ): + _add_column_if_missing(conn, "assistant_task_runs", name, statement) + _add_column_if_missing( + conn, + "assistant_tool_invocations", + "observed_at", + "ALTER TABLE assistant_tool_invocations ADD COLUMN observed_at DATETIME", + ) + _add_column_if_missing( + conn, + "assistant_tool_invocations", + "result_data", + "ALTER TABLE assistant_tool_invocations ADD COLUMN result_data JSON", + ) + _create_index_if_missing( + conn, + "ix_assistant_task_run_final_message", + "CREATE INDEX ix_assistant_task_run_final_message ON assistant_task_runs(final_message_id)", + ) + + MIGRATIONS: tuple[Migration, ...] = ( Migration(101, "agent_config_kind_and_visibility", _m101_agent_config_kind), Migration(102, "backfill_agent_kind_data", _m102_backfill_agent_kind), @@ -2040,6 +2072,7 @@ def _m127_assistant_trace_metrics(conn: Connection) -> None: Migration(125, "assistant_task_protocol", _m125_assistant_task_protocol), Migration(126, "assistant_task_events", _m126_assistant_task_events), Migration(127, "assistant_trace_metrics", _m127_assistant_trace_metrics), + Migration(128, "assistant_trusted_results", _m128_assistant_trusted_results), ) diff --git a/src/platform/persistence/models.py b/src/platform/persistence/models.py index 7c3026d3..b2890624 100644 --- a/src/platform/persistence/models.py +++ b/src/platform/persistence/models.py @@ -1147,6 +1147,7 @@ class AssistantTaskRun(Base): __table_args__ = ( Index("ix_assistant_task_run_conversation_created", "conversation_id", "created_at"), Index("ix_assistant_task_run_status_created", "status", "created_at"), + Index("ix_assistant_task_run_final_message", "final_message_id"), ) id = Column(Integer, primary_key=True, autoincrement=True) @@ -1170,6 +1171,8 @@ class AssistantTaskRun(Base): total_tokens = Column(Integer, nullable=False, default=0) cached_input_tokens = Column(Integer, nullable=False, default=0) reasoning_output_tokens = Column(Integer, nullable=False, default=0) + result_schema_version = Column(Integer, nullable=False, default=1) + result_data = Column(JSON, nullable=True) created_at = Column(DateTime, server_default=func.now()) started_at = Column(DateTime, nullable=True) finished_at = Column(DateTime, nullable=True) @@ -1228,6 +1231,8 @@ class AssistantToolInvocation(Base): status = Column(String, nullable=False, default="started") summary = Column(Text, nullable=False, default="") source_data = Column(JSON, default=[]) + result_data = Column(JSON, default={}) + observed_at = Column(DateTime, nullable=True) duration_ms = Column(Integer, nullable=False, default=0) attempt_count = Column(Integer, nullable=False, default=1) error_code = Column(String, nullable=True) diff --git a/tests/test_assistant_results.py b/tests/test_assistant_results.py new file mode 100644 index 00000000..c3cd1423 --- /dev/null +++ b/tests/test_assistant_results.py @@ -0,0 +1,169 @@ +"""Trusted assistant results remain deterministic and restorable.""" + +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import StaticPool + +from src.platform.persistence.database import Base +from src.platform.persistence.models import ChatConversation # noqa: F401 +from src.platform.tasking.contracts import TaskEventType, TaskStatus + + +def _invocation( + tool_name: str, + *, + call_id: str, + data: dict, + arguments: dict | None = None, + sources: list[dict] | None = None, + observed_at: datetime | None = None, + summary: str = "查询完成", +): + return SimpleNamespace( + call_id=call_id, + tool_name=tool_name, + status="completed", + summary=summary, + arguments=arguments or {}, + result_data=data, + source_data=sources or [], + observed_at=observed_at or datetime.now(UTC), + ) + + +def test_deterministic_result_uses_fields_and_flags_timepoint_discrepancies(): + from src.modules.assistant.result_builder import build_deterministic_assistant_result + + observed = datetime.now(UTC) + result = build_deterministic_assistant_result( + task_id=7, + answer="结论。", + language="zh-CN", + invocations=[ + _invocation( + "get_stock_quote", + call_id="quote-1", + data={"symbol": "600519", "market": "CN", "current_price": 105, "change_pct": 2.5}, + arguments={"symbol": "600519", "market": "CN"}, + sources=[{"name": "行情", "as_of": observed.isoformat()}], + observed_at=observed, + ), + _invocation( + "get_kline_summary", + call_id="kline-1", + data={"last_close": 100, "trend": "多头排列", "asof": "2026-09-28"}, + arguments={"symbol": "600519", "market": "CN"}, + sources=[{"name": "K 线", "as_of": "2026-09-28", "period_start": "2026-06-01", "period_end": "2026-09-28"}], + observed_at=observed, + ), + ], + ) + + assert any("最新价为 105" in fact.text for fact in result.facts) + assert any("收盘价 100" in fact.text and "多头排列" in fact.text for fact in result.facts) + assert any("时点不同" in risk for risk in result.risks) + assert any( + action.kind == "navigate" + and action.payload["path"] == "/portfolio?view=kline&symbol=600519&market=CN" + for action in result.next_actions + ) + + +def test_deterministic_result_reports_missing_fields_and_stale_evidence(): + from src.modules.assistant.result_builder import build_deterministic_assistant_result + + result = build_deterministic_assistant_result( + task_id=8, + answer="未取得完整数据。", + language="zh-CN", + invocations=[ + _invocation( + "get_stock_quote", + call_id="quote-1", + data={"symbol": "00700", "market": "HK", "current_price": None}, + arguments={"symbol": "00700", "market": "HK"}, + sources=[{"name": "行情", "as_of": (datetime.now(UTC) - timedelta(days=2)).isoformat()}], + ), + _invocation( + "get_stock_news", + call_id="news-1", + data={"symbol": "00700", "market": "HK", "items": []}, + arguments={"symbol": "00700", "market": "HK"}, + sources=[{"name": "新闻"}], + ), + ], + ) + + assert any("缺少最新价" in item for item in result.missing_data) + assert any("未检索到新闻" in item for item in result.missing_data) + assert any("已经过期" in item for item in result.risks) + + +def test_conversation_restores_legacy_result_and_trace_without_answer_payload(): + from src.modules.assistant.repository import AssistantRepository + from src.modules.assistant.service import AssistantService + + engine = create_engine( + "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool + ) + Base.metadata.create_all(engine) + session = sessionmaker(bind=engine)() + repository = AssistantRepository(session) + conversation = repository.create_conversation( + stock_symbol=None, stock_market=None, initial_context=None + ) + task = repository.create_task( + conversation_id=conversation.id, user_message_id=None, context={} + ) + repository.claim_task(task.id) + repository.append_task_event( + task.id, + TaskEventType.CONTEXT_PREPARED, + status=TaskStatus.RUNNING, + data={"compressed": False}, + ) + repository.record_tool_started( + task.id, + call_id="quote-1", + tool_name="get_stock_quote", + arguments={"symbol": "600519", "market": "CN"}, + ) + repository.record_tool_completed( + task.id, + call_id="quote-1", + tool_name="get_stock_quote", + summary="贵州茅台最新价 100。", + result_data={}, + sources=[{"name": "历史行情"}], + observed_at=datetime.now(UTC), + ) + message = repository.complete_task_with_message( + task.id, conversation.id, "历史结论。", result_data=None + ) + assert message is not None + + service = AssistantService(repository) + service._report_language = lambda: "zh-CN" # type: ignore[method-assign] + restored = service.get_conversation(conversation.id) + assistant_message = next(item for item in restored.messages if item.role == "assistant") + + assert assistant_message.result is not None + assert assistant_message.result.facts[0].text == "贵州茅台最新价 100。" + assert assistant_message.trace is not None + assert [item["event"] for item in assistant_message.trace] == [ + "run_started", + "context_prepared", + "tool_call_start", + "tool_result", + "done", + ] + done = assistant_message.trace[-1]["data"] + assert "content" not in done and "result" not in done + + invocation = repository.list_task_tool_invocations(task.id)[0] + assert invocation.result_data == {} + session.close() + engine.dispose() diff --git a/tests/test_assistant_stream_api.py b/tests/test_assistant_stream_api.py index ea14da6b..ee6632ce 100644 --- a/tests/test_assistant_stream_api.py +++ b/tests/test_assistant_stream_api.py @@ -313,7 +313,16 @@ async def run(): }] == [ ("step_updated", {"step": 1, "status": "running"}), ("tool_call_start", {"name": "get_price_alerts", "arguments": {"limit": 20}}), - ("tool_result", {"name": "get_price_alerts", "ok": True, "preview": "找到 1 条提醒"}), + ( + "tool_result", + { + "name": "get_price_alerts", + "ok": True, + "preview": "找到 1 条提醒", + "sources": [], + "observed_at": None, + }, + ), ] diff --git a/tests/test_assistant_task_events.py b/tests/test_assistant_task_events.py index cc9b8d7f..a93af9c4 100644 --- a/tests/test_assistant_task_events.py +++ b/tests/test_assistant_task_events.py @@ -224,6 +224,8 @@ def test_task_snapshot_contains_usage_and_tool_timing_summary(): "duration_ms": 420, "attempt_count": 2, "error_code": None, + "sources": [], + "observed_at": None, } ] @@ -315,3 +317,32 @@ def test_m126_creates_event_store_idempotently(tmp_path): engine.dispose() assert "assistant_task_events" in tables + + +def test_m128_adds_trusted_result_columns_idempotently(tmp_path): + from src.platform.persistence.migrations import ( + _m122_assistant_task_snapshots, + _m128_assistant_trusted_results, + ) + + engine = create_engine(f"sqlite:///{tmp_path / 'trusted-results.db'}") + with engine.begin() as conn: + _m122_assistant_task_snapshots(conn) + _m128_assistant_trusted_results(conn) + _m128_assistant_trusted_results(conn) + task_columns = { + row[1] for row in conn.execute(text("PRAGMA table_info(assistant_task_runs)")) + } + invocation_columns = { + row[1] + for row in conn.execute(text("PRAGMA table_info(assistant_tool_invocations)")) + } + indexes = { + row[1] + for row in conn.execute(text("PRAGMA index_list(assistant_task_runs)")) + } + engine.dispose() + + assert {"result_schema_version", "result_data"} <= task_columns + assert {"observed_at", "result_data"} <= invocation_columns + assert "ix_assistant_task_run_final_message" in indexes diff --git a/tests/test_assistant_task_repository.py b/tests/test_assistant_task_repository.py index 5a0106ae..c04670c4 100644 --- a/tests/test_assistant_task_repository.py +++ b/tests/test_assistant_task_repository.py @@ -31,6 +31,8 @@ def test_task_snapshot_contains_completed_tool_after_stream_expiry(): "duration_ms": 0, "attempt_count": 1, "error_code": None, + "sources": [], + "observed_at": None, }] session.close()