From b1427b57f4e74c0bc64142333f95be2b5d45fa84 Mon Sep 17 00:00:00 2001 From: 1012839419a-alt <268505792+1012839419a-alt@users.noreply.github.com> Date: Thu, 6 Aug 2026 10:22:51 +0800 Subject: [PATCH 1/4] add observable agent execution experience --- backend/api/v1/chat.py | 184 ++++++++++++++++- frontend/app/(app)/operations-agents/page.tsx | 11 +- .../components/shell/global-agent-dock.tsx | 186 ++++++++++++++++-- .../agent-execution-experience/brief.md | 28 +++ .../agent-execution-experience/design.md | 22 +++ .../agent-execution-experience/motion.md | 5 + .../changes/agent-execution-experience/qa.md | 14 ++ .../agent-execution-experience/tasks.md | 11 ++ 8 files changed, 446 insertions(+), 15 deletions(-) create mode 100644 openspec/changes/agent-execution-experience/brief.md create mode 100644 openspec/changes/agent-execution-experience/design.md create mode 100644 openspec/changes/agent-execution-experience/motion.md create mode 100644 openspec/changes/agent-execution-experience/qa.md create mode 100644 openspec/changes/agent-execution-experience/tasks.md diff --git a/backend/api/v1/chat.py b/backend/api/v1/chat.py index ee4f6368..edfe922f 100644 --- a/backend/api/v1/chat.py +++ b/backend/api/v1/chat.py @@ -11,12 +11,15 @@ v1 薄闭环: 唯一写动作 = 启停 source。验证通后按同模式扩 trigger_task / update_schedule。 """ +import asyncio import json import logging import re -from typing import Any, Literal, Optional +from contextvars import ContextVar +from typing import Any, Awaitable, Callable, Literal, Optional from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi.responses import StreamingResponse from pydantic import BaseModel from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession @@ -35,6 +38,40 @@ MAX_TOOL_STEPS = 5 +ActivitySink = Callable[[dict[str, Any]], Awaitable[None]] +_activity_sink: ContextVar[ActivitySink | None] = ContextVar("chat_activity_sink", default=None) + +_PUBLIC_TOOL_LABELS = { + "list_sources": ("检查数据源", "数据源"), + "list_schedules": ("检查调度计划", "调度计划"), + "list_tasks": ("检查最近任务", "采集任务"), + "list_providers": ("检查模型连接", "模型提供商"), + "toggle_source": ("变更数据源状态", "数据源"), + "trigger_task": ("启动采集任务", "数据源"), + "update_schedule": ("更新调度计划", "调度计划"), + "update_provider": ("更新模型配置", "模型提供商"), +} + + +async def _emit_activity(event_type: str, label: str, detail: str, **extra: Any) -> None: + sink = _activity_sink.get() + if sink is not None: + await sink({"type": event_type, "label": label, "detail": detail, **extra}) + + +def _tool_public_description(name: str, args: dict[str, Any]) -> tuple[str, str, str | None]: + label, target_type = _PUBLIC_TOOL_LABELS.get(name, ("执行操作", "系统对象")) + target_id = next((str(args[key]) for key in ("source_id", "schedule_id", "provider_id") if args.get(key)), None) + return label, target_type, target_id + + +def _result_public_summary(result: Any) -> str: + if isinstance(result, list): + return f"找到 {len(result)} 项可用信息" + if isinstance(result, dict) and result.get("error"): + return "未能读取目标信息" + return "已读取目标信息" + SYSTEM_PROMPT = """你是 opencli-admin 的全局操作助手。用户可能位于任意产品页面。\ 你的职责: 根据当前页面和对象上下文解释系统状态，并在已有工具覆盖范围内按用户意图查询或修改后端配置。 @@ -347,9 +384,21 @@ async def chat( identity: RequestIdentity | None = Depends(_optional_request_identity), db: AsyncSession = Depends(get_db), ) -> ApiResponse: + await _emit_activity( + "phase.changed", + "理解目标", + "正在结合当前页面、工作区和选中对象理解请求。", + state="completed", + ) provider = await _pick_provider(db, body.provider_id) client = await _build_client(provider) model = provider.default_model or "gpt-4o-mini" + await _emit_activity( + "phase.changed", + "制定执行路径", + "已选择可用模型，正在判断需要读取的信息和可能的操作。", + state="active", + ) system = SYSTEM_PROMPT if body.context: @@ -362,6 +411,12 @@ async def chat( messages += [{"role": m.role, "content": m.content} for m in body.messages] for _step in range(MAX_TOOL_STEPS): + await _emit_activity( + "phase.changed", + "分析当前状态", + "正在根据已获得的信息决定下一步。", + state="active", + ) try: response = await client.chat.completions.create( model=model, messages=messages, tools=TOOLS, tool_choice="auto" @@ -374,12 +429,26 @@ async def chat( tool_calls = msg.tool_calls or [] if not tool_calls: + await _emit_activity( + "run.completed", + "处理完成", + "已生成基于本次执行信息的结果摘要。", + state="completed", + ) return ApiResponse.ok(ChatReply(type="message", content=msg.content or "")) # 写工具命中 → 立即返回 proposal (不执行, 不继续推理) for tc in tool_calls: if tc.function.name in WRITE_TOOLS: args = _safe_json(tc.function.arguments) + label, target_type, target_id = _tool_public_description(tc.function.name, args) + await _emit_activity( + "tool.completed", + label, + "已定位目标并准备变更方案。", + state="completed", + target={"type": target_type, "id": target_id}, + ) proposal = await _build_proposal( db, tc.function.name, @@ -387,6 +456,13 @@ async def chat( identity=_require_write_identity(identity), workspace_id=_workspace_id(body.context), ) + await _emit_activity( + "approval.required", + "等待确认", + proposal.summary, + state="attention", + target={"type": target_type, "id": target_id}, + ) return ApiResponse.ok(ChatReply(type="proposal", proposal=proposal)) # 只读工具 → 执行, 喂回结果, 继续循环 @@ -405,7 +481,23 @@ async def chat( } ) for tc in tool_calls: - result = await _run_read_tool(db, tc.function.name, _safe_json(tc.function.arguments)) + args = _safe_json(tc.function.arguments) + label, target_type, target_id = _tool_public_description(tc.function.name, args) + await _emit_activity( + "tool.started", + label, + f"正在读取{target_type}的当前状态。", + state="active", + target={"type": target_type, "id": target_id}, + ) + result = await _run_read_tool(db, tc.function.name, args) + await _emit_activity( + "tool.completed", + label, + _result_public_summary(result), + state="completed", + target={"type": target_type, "id": target_id}, + ) messages.append( {"role": "tool", "tool_call_id": tc.id, "content": json.dumps(result, ensure_ascii=False)} ) @@ -413,6 +505,94 @@ async def chat( return ApiResponse.ok(ChatReply(type="message", content="(达到工具调用步数上限, 请换个说法再试)")) +@router.post("/stream") +async def chat_stream( + body: ChatRequest, + identity: RequestIdentity | None = Depends(_optional_request_identity), + db: AsyncSession = Depends(get_db), +) -> StreamingResponse: + """Stream public execution facts as NDJSON while the existing chat run executes. + + Events deliberately contain no model reasoning, raw tool arguments, credentials, or + unbounded tool results. The terminal ``reply`` event preserves the established ChatReply + contract so confirmation continues through the governed endpoint. + """ + + async def event_source(): + queue: asyncio.Queue[dict[str, Any] | None] = asyncio.Queue() + sequence = 0 + + async def emit(event: dict[str, Any]) -> None: + nonlocal sequence + sequence += 1 + await queue.put({"sequence": sequence, **event}) + + async def produce() -> None: + token = _activity_sink.set(emit) + try: + await emit( + { + "type": "run.started", + "label": "开始处理", + "detail": "已接收请求，正在建立执行上下文。", + "state": "active", + } + ) + response = await chat(body, identity, db) + await emit( + { + "type": "reply", + "label": "结果已就绪", + "detail": "本次处理已返回结果。", + "state": "completed", + "reply": response.data.model_dump(mode="json"), + } + ) + except HTTPException as exc: + await emit( + { + "type": "run.failed", + "label": "处理未完成", + "detail": str(exc.detail), + "state": "failed", + "status": exc.status_code, + "recovery": "检查连接或目标状态后重试。", + } + ) + except Exception: + logger.exception("chat stream failed") + await emit( + { + "type": "run.failed", + "label": "处理未完成", + "detail": "Agent 暂时无法完成这项任务。", + "state": "failed", + "status": 500, + "recovery": "稍后重试，或调整请求后继续。", + } + ) + finally: + _activity_sink.reset(token) + await queue.put(None) + + task = asyncio.create_task(produce()) + try: + while True: + event = await queue.get() + if event is None: + break + yield json.dumps(event, ensure_ascii=False) + "\n" + finally: + if not task.done(): + task.cancel() + + return StreamingResponse( + event_source(), + media_type="application/x-ndjson", + headers={"Cache-Control": "no-cache, no-transform", "X-Accel-Buffering": "no"}, + ) + + @router.post("/confirm", response_model=ApiResponse[dict]) async def confirm( body: ConfirmRequest, diff --git a/frontend/app/(app)/operations-agents/page.tsx b/frontend/app/(app)/operations-agents/page.tsx index 2476c970..07cb5fac 100644 --- a/frontend/app/(app)/operations-agents/page.tsx +++ b/frontend/app/(app)/operations-agents/page.tsx @@ -51,6 +51,15 @@ function parseJsonObject(value: string, label: string) { return parsed as Record } +function publicRunSummary(payload: Record | null) { + if (!payload) return null + const values = Object.entries(payload) + .filter(([, value]) => ['string', 'number', 'boolean'].includes(typeof value)) + .slice(0, 4) + .map(([key, value]) => `${key}: ${String(value)}`) + return values.length ? values.join(' · ') : '已生成结构化执行结果，可在运行记录中审计。' +} + function ContractEditor({ workspaceId, agent }: { workspaceId: string; agent: OperationsAgent }) { const draft = useOperationsAgentDraft(workspaceId, agent.id) const versions = useOperationsAgentVersions(workspaceId, agent.id) @@ -346,7 +355,7 @@ export default function OperationsAgentsPage() {
SESSION OUTPUT
- {(activity.data ?? []).filter((run) => run.operations_agent_id === selectedAgent.id).length ?
{(activity.data ?? []).filter((run) => run.operations_agent_id === selectedAgent.id).map((run) =>
[{run.status}] {new Date(run.updated_at).toLocaleString()}
{run.trigger_type} → {run.target_resource_type}/{run.target_resource_id}
profile v{run.profile_version} · agent v{run.published_version}
{run.error_message ?
{run.error_message}
: null}{run.output_payload ? 
{JSON.stringify(run.output_payload, null, 2)}
: null}
)}
:

还没有会话输出

智能体收到任务后，这里会显示真实的 CLI 活动和运行状态。

} + {(activity.data ?? []).filter((run) => run.operations_agent_id === selectedAgent.id).length ?
{(activity.data ?? []).filter((run) => run.operations_agent_id === selectedAgent.id).map((run) =>
{run.status === 'queued' ? '等待执行' : run.status === 'running' ? '正在执行' : run.status === 'completed' ? '已完成' : run.status === 'paused' ? '等待确认' : run.status === 'cancelled' ? '已取消' : '执行失败'}

目标：{run.target_resource_type} · {run.target_resource_id}

{run.error_message ?
{run.error_message}

检查目标状态后可以重新启动。

: null}{publicRunSummary(run.output_payload) ?
结果：{publicRunSummary(run.output_payload)}
: null}
)}
:

还没有执行活动

智能体收到任务后，这里会显示目标、当前状态和结果摘要。

}
diff --git a/frontend/components/shell/global-agent-dock.tsx b/frontend/components/shell/global-agent-dock.tsx index 148599db..f9871908 100644 --- a/frontend/components/shell/global-agent-dock.tsx +++ b/frontend/components/shell/global-agent-dock.tsx @@ -1,7 +1,7 @@ 'use client' import { useQueryClient } from '@tanstack/react-query' -import { Bot, Check, Loader2, Send, ShieldCheck, X } from 'lucide-react' +import { Bot, Check, CircleAlert, CircleCheck, Clock3, Loader2, Monitor, RotateCcw, Send, ShieldCheck, Sparkles, X } from 'lucide-react' import { usePathname } from 'next/navigation' import { FormEvent, KeyboardEvent, useState } from 'react' @@ -16,7 +16,7 @@ import { } from '@/components/ui/sheet' import { Textarea } from '@/components/ui/textarea' import { apiClient } from '@/lib/api/client' -import type { ApiResponse } from '@/lib/api/types' +import { getApiAuthHeaders } from '@/lib/api/auth-headers' import { ROUTE_LABELS } from '@/lib/navigation' type AgentMessage = { @@ -40,6 +40,45 @@ type AgentReply = { proposal?: AgentProposal | null } +type ActivityState = 'active' | 'complete' | 'attention' + +type Activity = { + label: string + detail: string + state: ActivityState + target?: { type?: string; id?: string | null } +} + +type AgentRunEvent = { + sequence: number + type: string + label: string + detail: string + state?: 'active' | 'completed' | 'attention' | 'failed' + target?: { type?: string; id?: string | null } + recovery?: string + reply?: AgentReply +} + +function activityFromEvent(event: AgentRunEvent): Activity { + return { + label: event.label, + detail: event.recovery ? `${event.detail} ${event.recovery}` : event.detail, + state: event.state === 'completed' ? 'complete' : event.state === 'failed' || event.state === 'attention' ? 'attention' : 'active', + target: event.target, + } +} + +function activityForReply(reply: AgentReply): Activity[] { + if (reply.type === 'proposal' && reply.proposal) { + return [ + { label: '已定位操作对象', detail: reply.proposal.summary, state: 'complete' }, + { label: '等待你的确认', detail: '这是一次会改变软件状态的操作。确认后才会执行。', state: 'attention' }, + ] + } + return [{ label: '已完成处理', detail: '已基于当前可访问的数据生成结果。', state: 'complete' }] +} + export function GlobalAgentDock({ open, onOpenChange, @@ -55,6 +94,10 @@ export function GlobalAgentDock({ const [error, setError] = useState(null) const [sending, setSending] = useState(false) const [confirming, setConfirming] = useState(false) + const [goal, setGoal] = useState(null) + const [activities, setActivities] = useState([]) + const [lastFailedProposal, setLastFailedProposal] = useState(null) + const [showLiveSurface, setShowLiveSurface] = useState(false) async function sendMessage(event?: FormEvent) { event?.preventDefault() @@ -66,6 +109,11 @@ export function GlobalAgentDock({ setInput('') setError(null) setSending(true) + setGoal(content) + setActivities([ + { label: '理解你的目标', detail: '正在结合当前页面和选中对象梳理任务。', state: 'complete' }, + { label: '检查可用信息', detail: '正在判断是否需要读取数据或准备操作。', state: 'active' }, + ]) try { const searchParams = new URLSearchParams(window.location.search) const workspaceId = searchParams.get('workspace') @@ -76,7 +124,10 @@ export function GlobalAgentDock({ const sourceId = searchParams.get('source') ?? pathname.match(/^\/sources\/([^/]+)/)?.[1] ?? null - const response = await apiClient.post>('/chat', { + const response = await fetch('/api/v1/chat/stream', { + method: 'POST', + headers: { 'Content-Type': 'application/json', ...getApiAuthHeaders() }, + body: JSON.stringify({ messages: nextMessages, context: { surface: ROUTE_LABELS[pathname] ?? pathname, @@ -87,8 +138,37 @@ export function GlobalAgentDock({ workflow_id: workflowId, source_id: sourceId, }, + }), }) - const reply = response.data.data + if (!response.ok || !response.body) throw new Error(`Agent 请求失败（${response.status}）`) + + const reader = response.body.getReader() + const decoder = new TextDecoder() + let buffer = '' + let reply: AgentReply | null = null + let streamError: string | null = null + while (true) { + const { value, done } = await reader.read() + buffer += decoder.decode(value, { stream: !done }) + const lines = buffer.split('\n') + buffer = lines.pop() ?? '' + for (const line of lines) { + if (!line.trim()) continue + const runEvent = JSON.parse(line) as AgentRunEvent + if (runEvent.type === 'reply' && runEvent.reply) { + reply = runEvent.reply + } else { + setActivities((current) => { + const next = [...current.filter((item) => item.state !== 'active'), activityFromEvent(runEvent)] + return next.slice(-8) + }) + } + if (runEvent.type === 'run.failed') streamError = runEvent.detail + } + if (done) break + } + if (streamError) throw new Error(streamError) + if (!reply) throw new Error('Agent 执行结束但没有返回结果') if (reply.type === 'proposal' && reply.proposal) { setProposal(reply.proposal) } else { @@ -97,25 +177,40 @@ export function GlobalAgentDock({ { role: 'assistant', content: reply.content?.trim() || '没有返回内容。' }, ]) } + setActivities((current) => [...current.filter((item) => item.state !== 'active'), ...activityForReply(reply)].slice(-8)) } catch (reason) { - setError(reason instanceof Error ? reason.message : 'Agent 暂时不可用') + const message = reason instanceof Error ? reason.message : 'Agent 暂时不可用' + setError(message) + setActivities([ + { label: '暂时无法完成理解', detail: message, state: 'attention' }, + { label: '恢复方式', detail: '请检查模型连接后重试，或换一种说法继续。', state: 'attention' }, + ]) } finally { setSending(false) } } - async function confirmProposal() { - if (!proposal || confirming) return + async function confirmProposal(proposalToConfirm = proposal) { + if (!proposalToConfirm || confirming) return setError(null) setConfirming(true) + setLastFailedProposal(null) + setActivities([ + { label: '已获得你的确认', detail: proposalToConfirm.summary, state: 'complete' }, + { label: '正在执行操作', detail: '系统正在应用这项变更。', state: 'active' }, + ]) try { - await apiClient.post('/chat/confirm', { proposal }) + await apiClient.post('/chat/confirm', { proposal: proposalToConfirm }) setMessages((current) => [ ...current, - { role: 'assistant', content: `已执行：${proposal.summary}` }, + { role: 'assistant', content: `已完成：${proposalToConfirm.summary}` }, ]) setProposal(null) await queryClient.invalidateQueries() + setActivities([ + { label: '操作已完成', detail: proposalToConfirm.summary, state: 'complete' }, + { label: '界面已同步', detail: '已刷新相关数据；你现在看到的是最新状态。', state: 'complete' }, + ]) } catch (reason) { const status = reason instanceof Error && 'status' in reason ? reason.status : undefined const message = reason instanceof Error ? reason.message : '操作执行失败' @@ -124,6 +219,11 @@ export function GlobalAgentDock({ ? `提案已失效或目标已变化：${message}。请拒绝后重新发起。` : message, ) + setLastFailedProposal(proposalToConfirm) + setActivities([ + { label: '操作未完成', detail: message, state: 'attention' }, + { label: '可恢复', detail: '检查目标状态后，可重新执行或回到对话调整请求。', state: 'attention' }, + ]) } finally { setConfirming(false) } @@ -178,14 +278,66 @@ export function GlobalAgentDock({ Agent 正在处理
) : null} + {goal ? ( +
+
+ + 正在处理 +
+

目标：{goal}

+
    + {activities.map((activity, index) => { + const Icon = activity.state === 'complete' ? CircleCheck : activity.state === 'attention' ? CircleAlert : Clock3 + return ( +
  1. + +
    +

    {activity.label}

    +

    {activity.detail}

    + {activity.target?.type ? ( +

    + 对象：{activity.target.type}{activity.target.id ? ` · ${activity.target.id}` : ''} +

    + ) : null} +
    +
    2. + ) + })} +
+
+ ) : null} + {goal ? ( +
+
+
+

+ + 软件现场 +

+

查看内置浏览器正在发生的实际变化。

+
+ +
+ {showLiveSurface ? ( +