diff --git a/server/routers/chat_router.py b/server/routers/chat_router.py index 1187c31c..750defe7 100644 --- a/server/routers/chat_router.py +++ b/server/routers/chat_router.py @@ -1,5 +1,6 @@ import traceback import uuid +from typing import Any from fastapi import APIRouter, Body, Depends, HTTPException, Query, UploadFile, File from fastapi.responses import StreamingResponse @@ -502,19 +503,57 @@ async def update_chat_models(model_provider: str, model_names: list[str], curren async def resume_agent_chat( agent_id: str, thread_id: str = Body(...), - approved: bool = Body(...), + approved: bool | None = Body(None), + answer: dict | list | str | None = Body(None), config: dict = Body({}), current_user: User = Depends(get_required_user), db: AsyncSession = Depends(get_db), ): """恢复被人工审批中断的对话（需要登录）""" - logger.info(f"Resuming agent_id: {agent_id}, thread_id: {thread_id}, approved: {approved}") + def normalize_resume_input(raw_answer: Any, raw_approved: bool | None) -> Any: + if raw_answer is not None: + if isinstance(raw_answer, str): + normalized = raw_answer.strip() + if not normalized: + raise HTTPException(status_code=422, detail="answer 不能为空") + return normalized + + if isinstance(raw_answer, list): + if len(raw_answer) == 0: + raise HTTPException(status_code=422, detail="answer 不能为空") + return raw_answer + + if isinstance(raw_answer, dict): + if raw_answer.get("type") == "other": + text = raw_answer.get("text") + if not isinstance(text, str) or not text.strip(): + raise HTTPException(status_code=422, detail="other 文本不能为空") + return raw_answer + + raise HTTPException(status_code=422, detail="answer 类型不支持") + + if raw_approved is not None: + return "approve" if raw_approved else "reject" + + raise HTTPException(status_code=422, detail="approved 或 answer 至少提供一个") + + resume_input = normalize_resume_input(answer, approved) + + logger.info( + "Resuming agent_id: %s, thread_id: %s, approved: %s, answer_type: %s", + agent_id, + thread_id, + approved, + type(answer).__name__ if answer is not None else "None", + ) meta = { "agent_id": agent_id, "thread_id": thread_id, "user_id": current_user.id, "approved": approved, + "answer": answer, + "resume_input": resume_input, } if "request_id" not in meta or not meta.get("request_id"): meta["request_id"] = str(uuid.uuid4()) @@ -522,7 +561,7 @@ async def resume_agent_chat( stream_agent_resume( agent_id=agent_id, thread_id=thread_id, - approved=approved, + resume_input=resume_input, meta=meta, config=config, current_user=current_user, diff --git a/src/agents/common/toolkits/buildin/__init__.py b/src/agents/common/toolkits/buildin/__init__.py index f51d0a80..7459ccc3 100644 --- a/src/agents/common/toolkits/buildin/__init__.py +++ b/src/agents/common/toolkits/buildin/__init__.py @@ -1,7 +1,8 @@ # buildin 工具包 -from .tools import calculator, query_knowledge_graph, text_to_img_qwen_image +from .tools import calculator, query_knowledge_graph, text_to_img_qwen_image, ask_user_question __all__ = [ + "ask_user_question", "calculator", "query_knowledge_graph", "text_to_img_qwen_image", diff --git a/src/agents/common/toolkits/buildin/tools.py b/src/agents/common/toolkits/buildin/tools.py index ed23665a..466b1a62 100644 --- a/src/agents/common/toolkits/buildin/tools.py +++ b/src/agents/common/toolkits/buildin/tools.py @@ -68,6 +68,73 @@ def calculator(a: float, b: float, operation: str) -> float: raise +ASK_USER_QUESTION_DESCRIPTION = """ +在执行过程中，当你需要用户做决定或补充需求时，使用这个工具向用户提问。 + +适用场景： +1. 收集用户偏好或需求（例如风格、范围、优先级） +2. 澄清模糊指令（存在多种合理解释时） +3. 在实现过程中让用户选择方案方向 +4. 在有明显权衡时让用户做取舍 + +使用规范： +1. 问题应当简短、具体、可回答，避免开放式长问句 +2. options 提供 2-5 个有区分度的选项，每项包含 label 和 value +3. 若有推荐选项：把推荐项放在第一位，并在 label 末尾加 "(Recommended)" +4. 若需要多选：将 multi_select 设为 true +5. allow_other 通常保持 true，用户可通过 Other 输入自定义答案 + +注意事项： +1. 不要用这个工具询问“是否继续执行”“计划是否准备好”这类流程控制问题 +2. 不要在信息已充分、无需用户决策时滥用该工具 +3. 先基于现有上下文自行决策，只有关键不确定性时才提问 + +返回结果： +answer 可能是 string（单选）、list（多选）或 object（Other 文本）。 +""" + + +@tool( + category="buildin", + tags=["交互"], + display_name="向用户提问", + description=ASK_USER_QUESTION_DESCRIPTION, +) +def ask_user_question( + question: Annotated[str, "向用户展示的问题"], + options: Annotated[list[dict], "候选项列表，格式 [{label, value}]，推荐项请在 label 后追加 (Recommended)"], + multi_select: Annotated[bool, "是否允许多选"] = False, + allow_other: Annotated[bool, "是否允许用户输入 Other 自定义答案"] = True, +) -> dict: + """向用户发起问题并等待回答。""" + normalized_options: list[dict[str, str]] = [] + for item in options or []: + if not isinstance(item, dict): + continue + label = str(item.get("label") or item.get("value") or "").strip() + value = str(item.get("value") or item.get("label") or "").strip() + if label and value: + normalized_options.append({"label": label, "value": value}) + + interrupt_payload = { + "question": question, + "question_id": str(uuid.uuid4()), + "options": normalized_options, + "multi_select": multi_select, + "allow_other": allow_other, + "source": "ask_user_question", + } + answer = interrupt(interrupt_payload) + + return { + "question": question, + "question_id": interrupt_payload["question_id"], + "answer": answer, + "multi_select": multi_select, + "allow_other": allow_other, + } + + KG_QUERY_DESCRIPTION = """ 使用这个工具可以查询知识图谱中包含的三元组信息。 关键词（query），使用可能帮助回答这个问题的关键词进行查询，不要直接使用用户的原始输入去查询。 diff --git a/src/agents/common/toolkits/debug/tools.py b/src/agents/common/toolkits/debug/tools.py index af5ae0b2..c85ae9b0 100644 --- a/src/agents/common/toolkits/debug/tools.py +++ b/src/agents/common/toolkits/debug/tools.py @@ -22,7 +22,25 @@ def get_approved_user_goal( } # 触发人工审批 - is_approved = interrupt(interrupt_info) + interrupt_result = interrupt(interrupt_info) + + if isinstance(interrupt_result, bool): + is_approved = interrupt_result + elif isinstance(interrupt_result, str): + is_approved = interrupt_result.strip().lower() in {"approve", "approved", "true", "yes", "1"} + elif isinstance(interrupt_result, list): + lowered = {str(item).strip().lower() for item in interrupt_result} + is_approved = "approve" in lowered or "approved" in lowered + elif isinstance(interrupt_result, dict): + selected = interrupt_result.get("selected") + if isinstance(selected, list): + lowered = {str(item).strip().lower() for item in selected} + is_approved = "approve" in lowered or "approved" in lowered + else: + text = str(interrupt_result.get("text") or "").strip().lower() + is_approved = text in {"approve", "approved", "true", "yes", "1"} + else: + is_approved = bool(interrupt_result) # 返回审批结果 if is_approved: diff --git a/src/services/chat_stream_service.py b/src/services/chat_stream_service.py index dfb14a93..7539d50d 100644 --- a/src/services/chat_stream_service.py +++ b/src/services/chat_stream_service.py @@ -4,6 +4,7 @@ import traceback import uuid from collections.abc import AsyncIterator from datetime import UTC, datetime +from typing import Any from langchain.messages import AIMessage, AIMessageChunk, HumanMessage from langgraph.types import Command @@ -193,7 +194,7 @@ async def save_messages_from_langgraph_state( logger.error(traceback.format_exc()) -def _extract_interrupt_info(state) -> dict | None: +def _extract_interrupt_info(state) -> Any | None: """从 LangGraph state 中提取中断信息""" if hasattr(state, "tasks") and state.tasks: for task in state.tasks: @@ -207,12 +208,73 @@ def _extract_interrupt_info(state) -> dict | None: return None -def _get_interrupt_fields(info) -> tuple[str, str]: - """从中断信息中提取 question 和 operation""" - defaults = ("是否批准以下操作？", "需要人工审批的操作") +def _coerce_interrupt_payload(info: Any) -> dict: + """将 LangGraph interrupt 对象转换为 dict 结构。""" if isinstance(info, dict): - return info.get("question", defaults[0]), info.get("operation", defaults[1]) - return getattr(info, "question", defaults[0]), getattr(info, "operation", defaults[1]) + return info + + payload = getattr(info, "value", None) + if isinstance(payload, dict): + return payload + + question = getattr(info, "question", None) + operation = getattr(info, "operation", None) + result: dict[str, Any] = {} + if isinstance(question, str) and question.strip(): + result["question"] = question + if isinstance(operation, str) and operation.strip(): + result["operation"] = operation + return result + + +def _normalize_interrupt_options(raw_options: Any) -> list[dict[str, str]]: + if not isinstance(raw_options, list): + return [] + + options: list[dict[str, str]] = [] + for item in raw_options: + if isinstance(item, dict): + label = str(item.get("label") or item.get("value") or "").strip() + value = str(item.get("value") or item.get("label") or "").strip() + else: + label = str(item).strip() + value = label + if label and value: + options.append({"label": label, "value": value}) + return options + + +def _build_ask_user_question_payload(info: Any, thread_id: str) -> dict[str, Any]: + """将 interrupt 信息标准化为 ask_user_question_required 载荷。""" + payload = _coerce_interrupt_payload(info) + + question = str(payload.get("question") or "请选择一个选项").strip() + question_id = str(payload.get("question_id") or uuid.uuid4()) + source = str(payload.get("source") or payload.get("tool_name") or "interrupt") + multi_select = bool(payload.get("multi_select", False)) + allow_other = bool(payload.get("allow_other", True)) + operation = payload.get("operation") + + options = _normalize_interrupt_options(payload.get("options")) + if not options and isinstance(operation, str) and operation.strip(): + # 兼容旧版 get_approved_user_goal 的 interrupt 结构 + options = [ + {"label": "批准 (Recommended)", "value": "approve"}, + {"label": "拒绝", "value": "reject"}, + ] + source = "get_approved_user_goal" + allow_other = False + + return { + "question_id": question_id, + "question": question, + "options": options, + "multi_select": multi_select, + "allow_other": allow_other, + "source": source, + "operation": operation if isinstance(operation, str) else "", + "thread_id": thread_id, + } def _ensure_full_msg(full_msg: AIMessage | None, accumulated_content: list[str]) -> AIMessage | None: @@ -262,13 +324,9 @@ async def check_and_handle_interrupts( interrupt_info = _extract_interrupt_info(state) if interrupt_info: - question, operation = _get_interrupt_fields(interrupt_info) - meta["interrupt"] = { - "question": question, - "operation": operation, - "thread_id": thread_id, - } - yield make_chunk(status="interrupted", message=question, meta=meta) + question_payload = _build_ask_user_question_payload(interrupt_info, thread_id) + meta["interrupt"] = question_payload + yield make_chunk(status="ask_user_question_required", meta=meta, **question_payload) except Exception as e: logger.error(f"Error checking interrupts: {e}") @@ -357,6 +415,8 @@ async def stream_agent_chat( "agent_config_id": agent_config_id, "agent_config": agent_config, } + full_msg = None + accumulated_content: list[str] = [] try: conv_repo = ConversationRepository(db) @@ -489,7 +549,7 @@ async def stream_agent_resume( *, agent_id: str, thread_id: str, - approved: bool, + resume_input: Any, meta: dict, config: dict, current_user, @@ -515,10 +575,10 @@ async def stream_agent_resume( ) return - init_msg = {"type": "system", "content": f"Resume with approved: {approved}"} + init_msg = {"type": "system", "content": f"Resume with input: {resume_input}"} yield make_resume_chunk(status="init", meta=meta, msg=init_msg) - resume_command = Command(resume=approved) + resume_command = Command(resume=resume_input) graph = await agent.get_graph() user_id = str(current_user.id) diff --git a/src/services/run_worker.py b/src/services/run_worker.py index d6c13797..ebd6403f 100644 --- a/src/services/run_worker.py +++ b/src/services/run_worker.py @@ -275,6 +275,14 @@ async def process_agent_run(ctx, run_id: str): error_message=chunk.get("message"), ) terminal_set = True + elif status == "ask_user_question_required": + await mark_run_terminal( + run_id, + "interrupted", + error_type="ask_user_question_required", + error_message=chunk.get("question") or "需要用户回答问题", + ) + terminal_set = True if await run_ctx.is_cancelled(): raise asyncio.CancelledError(f"run {run_id} cancelled") diff --git a/web/src/components/AgentChatComponent.vue b/web/src/components/AgentChatComponent.vue index b0625afe..493cb395 100644 --- a/web/src/components/AgentChatComponent.vue +++ b/web/src/components/AgentChatComponent.vue @@ -112,8 +112,11 @@ :visible="approvalState.showModal" :question="approvalState.question" :operation="approvalState.operation" - @approve="handleApprove" - @reject="handleReject" + :options="approvalState.options" + :multi-select="approvalState.multiSelect" + :allow-other="approvalState.allowOther" + @submit="handleQuestionSubmit" + @cancel="handleQuestionCancel" />
@@ -1113,7 +1116,12 @@ const startRunStream = async (threadId, runId, afterSeq = '0') => { } } - if (event === 'finished' || event === 'error' || event === 'interrupted') { + if ( + event === 'finished' || + event === 'error' || + event === 'interrupted' || + event === 'ask_user_question_required' + ) { flushTypingQueueForThread(threadId) ts.isStreaming = false ts.activeRunId = null @@ -1536,10 +1544,10 @@ const handleSendOrStop = async (payload) => { } // ==================== 人工审批处理 ==================== -const handleApprovalWithStream = async (approved) => { +const handleApprovalWithStream = async (answer) => { const threadId = approvalState.threadId if (!threadId) { - message.error('无效的审批请求') + message.error('无效的提问请求') approvalState.showModal = false return } @@ -1553,11 +1561,7 @@ const handleApprovalWithStream = async (approved) => { try { // 使用审批 composable 处理审批 - const response = await handleApproval( - approved, - currentAgentId.value, - selectedAgentConfigId.value - ) + const response = await handleApproval(answer, currentAgentId.value, selectedAgentConfigId.value) if (!response) return // 如果 handleApproval 抛出错误，这里不会执行 @@ -1581,12 +1585,12 @@ const handleApprovalWithStream = async (approved) => { } } -const handleApprove = () => { - handleApprovalWithStream(true) +const handleQuestionSubmit = (answer) => { + handleApprovalWithStream(answer) } -const handleReject = () => { - handleApprovalWithStream(false) +const handleQuestionCancel = () => { + handleApprovalWithStream('reject') } // 处理示例问题点击 diff --git a/web/src/components/HumanApprovalModal.vue b/web/src/components/HumanApprovalModal.vue index 3336121f..7a4f8992 100644 --- a/web/src/components/HumanApprovalModal.vue +++ b/web/src/components/HumanApprovalModal.vue @@ -6,19 +6,49 @@

{{ question }}

-
+
操作： {{ operation }}
+ +
+ +
+ +
+ +
- - + +
@@ -30,47 +60,114 @@ @@ -112,6 +209,7 @@ const handleReject = () => { line-height: 1.5; display: flex; gap: 6px; + margin-bottom: 10px; } .approval-operation .label { @@ -125,6 +223,42 @@ const handleReject = () => { word-break: break-word; } +.question-options { + display: flex; + flex-direction: column; + gap: 8px; +} + +.option-item { + display: flex; + align-items: center; + gap: 8px; + color: var(--gray-800); + font-size: 14px; +} + +.option-item .recommended { + color: var(--main-color); + font-weight: 600; +} + +.other-input { + margin-top: 10px; +} + +.other-input input { + width: 100%; + border: 1px solid var(--gray-300); + border-radius: 6px; + padding: 8px 10px; + font-size: 13px; + outline: none; +} + +.other-input input:focus { + border-color: var(--main-color); +} + .approval-actions { display: flex; gap: 10px; @@ -140,10 +274,6 @@ const handleReject = () => { font-weight: 500; cursor: pointer; transition: all 0.2s ease; - display: flex; - align-items: center; - justify-content: center; - gap: 6px; } .btn:disabled { @@ -167,7 +297,6 @@ const handleReject = () => { .btn-approve:hover:not(:disabled) { background: var(--main-700); - box-shadow: 0 2px 6px rgba(59, 130, 246, 0.25); } .approval-processing { @@ -197,7 +326,6 @@ const handleReject = () => { } } -/* 滑入滑出动画 */ .slide-up-enter-active, .slide-up-leave-active { transition: all 0.25s ease; diff --git a/web/src/composables/useAgentStreamHandler.js b/web/src/composables/useAgentStreamHandler.js index 0941c5bd..f3385ae6 100644 --- a/web/src/composables/useAgentStreamHandler.js +++ b/web/src/composables/useAgentStreamHandler.js @@ -110,6 +110,7 @@ export function useAgentStreamHandler({ } return true + case 'ask_user_question_required': case 'human_approval_required': console.log(`${debugPrefix}[approval_required]`, { threadId, diff --git a/web/src/composables/useApproval.js b/web/src/composables/useApproval.js index 5c6d252e..52a87664 100644 --- a/web/src/composables/useApproval.js +++ b/web/src/composables/useApproval.js @@ -3,21 +3,79 @@ import { message } from 'ant-design-vue' import { handleChatError } from '@/utils/errorHandler' import { agentApi } from '@/apis' +const normalizeOptions = (rawOptions) => { + if (!Array.isArray(rawOptions)) return [] + return rawOptions + .map((item) => { + if (item && typeof item === 'object') { + const label = String(item.label || item.value || '').trim() + const value = String(item.value || item.label || '').trim() + return label && value ? { label, value } : null + } + const text = String(item || '').trim() + return text ? { label: text, value: text } : null + }) + .filter(Boolean) +} + +const toLegacyApprovalOptions = () => [ + { label: '批准 (Recommended)', value: 'approve' }, + { label: '拒绝', value: 'reject' } +] + +const extractQuestionPayload = (chunk) => { + const interruptInfo = chunk?.interrupt_info || {} + const rawOptions = + chunk?.options || interruptInfo?.options || (interruptInfo?.operation ? toLegacyApprovalOptions() : []) + const options = normalizeOptions(rawOptions) + const operation = chunk?.operation || interruptInfo?.operation || '' + + const source = chunk?.source || interruptInfo?.source || (operation ? 'get_approved_user_goal' : 'interrupt') + const multiSelect = Boolean(chunk?.multi_select ?? interruptInfo?.multi_select ?? false) + let allowOther = Boolean(chunk?.allow_other ?? interruptInfo?.allow_other ?? true) + const questionId = chunk?.question_id || interruptInfo?.question_id || '' + const question = chunk?.question || interruptInfo?.question || '请选择一个选项' + const legacyMode = source === 'get_approved_user_goal' && options.length === 2 + if (source === 'get_approved_user_goal') { + allowOther = false + } + + return { + questionId, + question, + options, + multiSelect, + allowOther, + source, + operation, + legacyMode + } +} + +const inferApprovedFromAnswer = (answer) => { + if (answer === 'approve') return true + if (answer === 'reject') return false + return null +} + export function useApproval({ getThreadState, resetOnGoingConv, fetchThreadMessages }) { - // 审批状态 const approvalState = reactive({ showModal: false, + questionId: '', question: '', operation: '', - threadId: null, - interruptInfo: null + options: [], + multiSelect: false, + allowOther: true, + source: '', + legacyMode: false, + threadId: null }) - // 处理审批逻辑 - const handleApproval = async (approved, currentAgentId, agentConfigId = null) => { + const handleApproval = async (answer, currentAgentId, agentConfigId = null) => { const threadId = approvalState.threadId if (!threadId) { - message.error('无效的审批请求') + message.error('无效的提问请求') approvalState.showModal = false return } @@ -29,94 +87,93 @@ export function useApproval({ getThreadState, resetOnGoingConv, fetchThreadMessa return } - // 关闭弹窗 approvalState.showModal = false - // 清理旧的流式控制器（如果存在） if (threadState.streamAbortController) { threadState.streamAbortController.abort() threadState.streamAbortController = null } - // 标记为处理中 threadState.isStreaming = true resetOnGoingConv(threadId) threadState.streamAbortController = new AbortController() - console.log('🔄 [APPROVAL] Starting resume process:', { approved, threadId, currentAgentId }) + const approved = inferApprovedFromAnswer(answer) + const requestBody = { + thread_id: threadId, + answer, + config: agentConfigId ? { agent_config_id: agentConfigId } : {} + } + if (approved !== null) { + requestBody.approved = approved + } try { - // 调用恢复接口 - const response = await agentApi.resumeAgentChat( - currentAgentId, - { - thread_id: threadId, - approved: approved, - config: agentConfigId ? { agent_config_id: agentConfigId } : {} - }, - { - signal: threadState.streamAbortController?.signal - } - ) - - console.log('🔄 [APPROVAL] Resume API response received') + const response = await agentApi.resumeAgentChat(currentAgentId, requestBody, { + signal: threadState.streamAbortController?.signal + }) if (!response.ok) { const errorText = await response.text() - console.error('Resume API error:', response.status, errorText) throw new Error(`HTTP error! status: ${response.status}, details: ${errorText}`) } - console.log('🔄 [APPROVAL] Resume API successful, returning response for stream processing') - return response // 返回响应供调用方处理流式数据 + return response } catch (error) { - console.error('❌ [APPROVAL] Resume failed:', error) if (error.name !== 'AbortError') { handleChatError(error, 'resume') message.error(`恢复对话失败: ${error.message || '未知错误'}`) } - // 重置状态 - 只在错误时重置 threadState.isStreaming = false threadState.streamAbortController = null - throw error // 重新抛出错误让调用方处理 + throw error } - // 移除 finally 块 - 让组件管理流式状态的生命周期 } - // 在流式处理中处理审批请求 const processApprovalInStream = (chunk, threadId, currentAgentId) => { - if (chunk.status !== 'human_approval_required') { + if (chunk.status !== 'ask_user_question_required' && chunk.status !== 'human_approval_required') { return false } - const { interrupt_info } = chunk const threadState = getThreadState(threadId) - if (!threadState) return false - // 停止显示"处理中"状态，让用户可以看到并操作审批弹窗 + const payload = extractQuestionPayload(chunk) + if (!payload.options.length) { + payload.options = toLegacyApprovalOptions() + payload.legacyMode = true + payload.allowOther = false + } + threadState.isStreaming = false - // 显示审批弹窗 approvalState.showModal = true - approvalState.question = interrupt_info?.question || '是否批准以下操作？' - approvalState.operation = interrupt_info?.operation || '未知操作' + approvalState.questionId = payload.questionId + approvalState.question = payload.question + approvalState.operation = payload.operation + approvalState.options = payload.options + approvalState.multiSelect = payload.multiSelect + approvalState.allowOther = payload.allowOther + approvalState.source = payload.source + approvalState.legacyMode = payload.legacyMode approvalState.threadId = chunk.thread_id || threadId - approvalState.interruptInfo = interrupt_info - // 刷新消息历史显示已执行的部分 - fetchThreadMessages({ agentId: currentAgentId, threadId: threadId }) + fetchThreadMessages({ agentId: currentAgentId, threadId }) - return true // 表示已处理审批请求，应停止流式处理 + return true } - // 重置审批状态 const resetApprovalState = () => { approvalState.showModal = false + approvalState.questionId = '' approvalState.question = '' approvalState.operation = '' + approvalState.options = [] + approvalState.multiSelect = false + approvalState.allowOther = true + approvalState.source = '' + approvalState.legacyMode = false approvalState.threadId = null - approvalState.interruptInfo = null } return {