diff --git a/agent_platform/runtime/app.py b/agent_platform/runtime/app.py index 61e70e8..51607a5 100644 --- a/agent_platform/runtime/app.py +++ b/agent_platform/runtime/app.py @@ -22,6 +22,18 @@ from agent_platform.runtime.schemas import ChatCompletionRequest from agent_platform.runtime.tools import ToolRegistry from agent_platform.store import RuntimeStore +OPENWEBUI_BACKGROUND_TASKS = { + "title_generation", + "follow_up_generation", + "tags_generation", + "emoji_generation", + "query_generation", + "image_prompt_generation", + "autocomplete_generation", + "function_calling", + "moa_response_generation", +} + def _sse_chunk(model: str, content: str = "", finish_reason: str | None = None) -> bytes: payload = { @@ -145,15 +157,7 @@ def create_app( except ValueError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc - stable_chat_id = (chat_id or message_id or f"ephemeral-{uuid.uuid4().hex}").strip() - selected_mode = await app.state.store.select_mode(identity.user_id, chat_id or "", spec.mode) - if selected_mode == "work" and spec.mode == "chat": - raise HTTPException( - status_code=status.HTTP_409_CONFLICT, - detail="This conversation has been upgraded to Work and cannot return to Chat.", - ) - - if spec.mode == "chat": + async def forward_direct(): payload = body.model_dump(exclude_none=True) payload["model"] = spec.provider_model response = await app.state.provider.forward(payload) @@ -183,6 +187,21 @@ def create_app( headers=headers, ) + background_task = str((body.metadata or {}).get("task", "")).strip() + if background_task in OPENWEBUI_BACKGROUND_TASKS: + return await forward_direct() + + stable_chat_id = (chat_id or message_id or f"ephemeral-{uuid.uuid4().hex}").strip() + selected_mode = await app.state.store.select_mode(identity.user_id, chat_id or "", spec.mode) + if selected_mode == "work" and spec.mode == "chat": + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="This conversation has been upgraded to Work and cannot return to Chat.", + ) + + if spec.mode == "chat": + return await forward_direct() + messages = [message.model_dump(exclude_none=True) for message in body.messages] if not body.stream: diff --git a/agent_platform/runtime/loop.py b/agent_platform/runtime/loop.py index 7819149..14d118c 100644 --- a/agent_platform/runtime/loop.py +++ b/agent_platform/runtime/loop.py @@ -32,6 +32,9 @@ relevant verification, and report concrete results. Use tools instead of inventi command output. A claim that a file was created, changed, executed, or tested must be backed by a successful tool result from this run. For scripts, reports, code, and other deliverables, create real files under /workspace, inspect them after writing, run relevant checks, and cite their exact paths. +Write source files with real line breaks rather than literal `\\n` escape sequences. If any command +fails, inspect the failure, repair the underlying problem, and successfully rerun the relevant check; +never replace failed output with invented numbers, placeholders, or a hand-written success report. Never use an interactive command such as bare `python3`, `bash`, or `node`; provide a complete non-interactive command. Do not use `echo` merely to simulate progress. Keep the plan current. Use background processes for servers or long jobs. Delegate bounded independent investigations when it @@ -59,8 +62,15 @@ ACTION_REQUEST = re.compile( r"\b(?:run|execute|test|inspect|analy[sz]e|compare|debug|deploy)\b)", re.IGNORECASE, ) +EXECUTION_REQUEST = re.compile( + r"(运行|执行|测试|脚本|程序|代码|比较|对比|基准|" + r"\b(?:run|execute|test|script|program|code|compare|benchmark)\b)", + re.IGNORECASE, +) MUTATION_TOOLS = {"write_file", "apply_patch"} -VERIFICATION_TOOLS = {"read_file", "list_files", "search_files", "git_status", "git_diff", "exec", "poll_process"} +VERIFICATION_TOOLS = {"read_file", "git_status", "git_diff", "exec", "poll_process"} +EXECUTION_TOOLS = {"exec", "start_process", "poll_process"} +MAX_CONSECUTIVE_CHECKPOINT_REJECTIONS = 3 def latest_user_text(messages: list[dict[str, Any]]) -> str: @@ -78,19 +88,49 @@ def latest_user_text(messages: list[dict[str, Any]]) -> str: return "" -def completion_failure(artifact_required: bool, mutation_count: int, verified_mutation: int) -> str | None: +def completion_failure( + artifact_required: bool, + mutation_count: int, + verified_mutation: int, + action_required: bool = False, + execution_required: bool = False, + successful_executions: int = 0, + unresolved_execution_failure: str | None = None, +) -> str | None: + reasons: list[str] = [] if artifact_required and mutation_count == 0: - return ( + reasons.append( "The user requested a concrete deliverable, but this run has no successful file-writing tool call. " - "Do not claim completion. Create the requested files with write_file or apply_patch, then inspect " - "and verify them with tools." + "Create the requested files with write_file or apply_patch." ) - if artifact_required and verified_mutation < mutation_count: - return ( + elif artifact_required and verified_mutation < mutation_count: + reasons.append( "Files changed, but no successful verification occurred after the latest change. Inspect the actual " - "files and run the relevant check before claiming completion." + "files and run the relevant check." ) - return None + if action_required and unresolved_execution_failure: + reasons.append( + "A command or process check failed and has not been followed by a successful execution. " + f"Repair and rerun it. Last failure: {unresolved_execution_failure}" + ) + elif execution_required and successful_executions == 0: + reasons.append( + "The task requires a script, code, test, benchmark, or comparison, but no command or process " + "completed successfully. Run the relevant verification and use its real output." + ) + if not reasons: + return None + return " Do not claim completion. ".join(reasons) + + +def recovery_message(failure: str) -> str: + return ( + "Your attempted checkpoint was rejected by the execution policy. Continue the original task now. " + "You MUST call the appropriate tools in this response instead of repeating or paraphrasing a final answer. " + f"Evidence gap: {failure} " + "If a command failed, inspect and repair the real files, rerun the command successfully, then inspect the " + "resulting deliverables before finishing." + ) @dataclass(slots=True) @@ -147,7 +187,7 @@ class AgentLoop: "run.created", { "model_tier": spec.strength, - "strategy_version": "work-loop-v1", + "strategy_version": "work-loop-v2", "scheduler_version": "safe-parallel-v1", "context_policy": "recent-visible-v1", }, @@ -200,9 +240,13 @@ class AgentLoop: request_text = latest_user_text(messages) artifact_required = depth == 0 and bool(ARTIFACT_REQUEST.search(request_text)) action_required = depth == 0 and bool(ACTION_REQUEST.search(request_text)) + execution_required = depth == 0 and bool(EXECUTION_REQUEST.search(request_text)) mutation_count = 0 verified_mutation = -1 successful_tools = 0 + successful_executions = 0 + unresolved_execution_failure: str | None = None + consecutive_checkpoint_rejections = 0 for iteration in range(max_iterations): await recorder.emit( "model.requested", @@ -228,13 +272,26 @@ class AgentLoop: tool_calls = message.get("tool_calls") or [] if not tool_calls: candidate = str(message.get("content") or "").strip() - failure = completion_failure(artifact_required, mutation_count, verified_mutation) + failure = completion_failure( + artifact_required, + mutation_count, + verified_mutation, + action_required, + execution_required, + successful_executions, + unresolved_execution_failure, + ) if not failure and action_required and successful_tools == 0: failure = ( "The user requested an executed or inspected result, but no tool completed successfully. " "Use the appropriate workspace tools and return evidence instead of an unverified answer." ) - if failure and iteration + 1 < max_iterations: + consecutive_checkpoint_rejections += 1 + if ( + failure + and iteration + 1 < max_iterations + and consecutive_checkpoint_rejections < MAX_CONSECUTIVE_CHECKPOINT_REJECTIONS + ): await recorder.emit( "completion.rejected", {"reason": failure, "iteration": iteration + 1, "depth": depth}, @@ -242,7 +299,7 @@ class AgentLoop: messages.extend( [ {"role": "assistant", "content": candidate}, - {"role": "system", "content": failure}, + {"role": "user", "content": recovery_message(failure)}, ] ) continue @@ -257,6 +314,7 @@ class AgentLoop: ) return candidate + consecutive_checkpoint_rejections = 0 assistant_message = { "role": "assistant", "content": message.get("content"), @@ -279,6 +337,11 @@ class AgentLoop: mutation_count += 1 elif name in VERIFICATION_TOOLS and mutation_count: verified_mutation = mutation_count + if name in EXECUTION_TOOLS: + successful_executions += 1 + unresolved_execution_failure = None + elif name in EXECUTION_TOOLS: + unresolved_execution_failure = f"{name}: {public_tool_summary(result, 1000)}" messages.append( { "role": "tool", @@ -299,7 +362,15 @@ class AgentLoop: ) response = await self.provider.complete(model=spec.provider_model, messages=messages, tools=None) answer = str(response["choices"][0].get("message", {}).get("content") or "").strip() - if completion_failure(artifact_required, mutation_count, verified_mutation): + if completion_failure( + artifact_required, + mutation_count, + verified_mutation, + action_required, + execution_required, + successful_executions, + unresolved_execution_failure, + ): return ( "未完成：工具迭代已用尽，但交付物没有形成完整的“写入后验证”证据链。" "我不会声称文件已经完成；工作区中已有内容可从“工作区文件”查看。" diff --git a/agent_platform/runtime/schemas.py b/agent_platform/runtime/schemas.py index a4e2255..61916de 100644 --- a/agent_platform/runtime/schemas.py +++ b/agent_platform/runtime/schemas.py @@ -23,6 +23,7 @@ class ChatCompletionRequest(BaseModel): stream: bool = False tools: list[dict[str, Any]] | None = None tool_choice: Any = None + metadata: dict[str, Any] | None = None class PlanItem(BaseModel): diff --git a/compose.yaml b/compose.yaml index 86bdb69..869fcca 100644 --- a/compose.yaml +++ b/compose.yaml @@ -39,6 +39,11 @@ services: TASK_MODEL: chat-light ENABLE_TITLE_GENERATION: "false" ENABLE_FOLLOW_UP_GENERATION: "false" + ENABLE_TAGS_GENERATION: "false" + ENABLE_SEARCH_QUERY_GENERATION: "false" + ENABLE_RETRIEVAL_QUERY_GENERATION: "false" + ENABLE_AUTOCOMPLETE_GENERATION: "false" + ENABLE_IMAGE_PROMPT_GENERATION: "false" ENABLE_OLLAMA_API: "false" ENABLE_DIRECT_CONNECTIONS: "false" ENABLE_PERSISTENT_CONFIG: "false" diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index fd9b882..75af2e6 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -39,6 +39,11 @@ services: TASK_MODEL: chat-light ENABLE_TITLE_GENERATION: "false" ENABLE_FOLLOW_UP_GENERATION: "false" + ENABLE_TAGS_GENERATION: "false" + ENABLE_SEARCH_QUERY_GENERATION: "false" + ENABLE_RETRIEVAL_QUERY_GENERATION: "false" + ENABLE_AUTOCOMPLETE_GENERATION: "false" + ENABLE_IMAGE_PROMPT_GENERATION: "false" ENABLE_OLLAMA_API: "false" ENABLE_DIRECT_CONNECTIONS: "false" ENABLE_PERSISTENT_CONFIG: "false" diff --git a/tests/test_agent_loop.py b/tests/test_agent_loop.py index ab789ab..07b67c7 100644 --- a/tests/test_agent_loop.py +++ b/tests/test_agent_loop.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import copy import json from typing import Any @@ -25,7 +26,7 @@ class ScriptedProvider: self.requests = [] async def complete(self, **kwargs): - self.requests.append(kwargs) + self.requests.append(copy.deepcopy(kwargs)) return self.responses.pop(0) @@ -203,7 +204,7 @@ async def test_artifact_completion_is_rejected_until_written_and_verified(settin try: answer = await AgentLoop(provider, registry, store, max_tool_output_chars=10_000).run( spec=get_model_spec("work-light"), - messages=[{"role": "user", "content": "写一个脚本并生成报告"}], + messages=[{"role": "user", "content": "写一个报告"}], identity=UserIdentity("u1", "", "", "user"), raw_user_jwt="jwt", chat_id="c1", @@ -212,6 +213,191 @@ async def test_artifact_completion_is_rejected_until_written_and_verified(settin assert answer == "已完成 report.md" assert [name for name, _ in registry.calls] == ["write_file", "read_file"] assert "completion.rejected" in events + assert provider.requests[1]["messages"][-1]["role"] == "user" + assert "MUST call" in provider.requests[1]["messages"][-1]["content"] + finally: + await store.close() + + +class FailingExecutionRegistry(RecordingRegistry): + def __init__(self) -> None: + super().__init__() + self.exec_calls = 0 + + async def execute(self, name, arguments, context): + self.calls.append((name, arguments)) + if name == "exec": + self.exec_calls += 1 + if self.exec_calls == 1: + return {"ok": False, "error": "SyntaxError: invalid source"} + return {"ok": True, "output": f"{name} complete"} + + +async def test_failed_execution_must_be_repaired_before_completion(settings) -> None: + write = { + "choices": [ + { + "message": { + "role": "assistant", + "tool_calls": [tool_call("1", "write_file", {"path": "sort.py", "content": "broken"})], + }, + "finish_reason": "tool_calls", + } + ] + } + read = { + "choices": [ + { + "message": { + "role": "assistant", + "tool_calls": [tool_call("2", "read_file", {"path": "sort.py"})], + }, + "finish_reason": "tool_calls", + } + ] + } + failed_exec = { + "choices": [ + { + "message": { + "role": "assistant", + "tool_calls": [tool_call("3", "exec", {"command": "python3 sort.py"})], + }, + "finish_reason": "tool_calls", + } + ] + } + premature = {"choices": [{"message": {"role": "assistant", "content": "已完成"}, "finish_reason": "stop"}]} + repaired_exec = { + "choices": [ + { + "message": { + "role": "assistant", + "tool_calls": [tool_call("4", "exec", {"command": "python3 sort.py"})], + }, + "finish_reason": "tool_calls", + } + ] + } + final = { + "choices": [{"message": {"role": "assistant", "content": "脚本运行和报告验证完成"}, "finish_reason": "stop"}] + } + provider = ScriptedProvider([write, read, failed_exec, premature, repaired_exec, final]) + registry = FailingExecutionRegistry() + store = RuntimeStore(settings.database_url) + await store.initialize() + events = [] + + async def callback(event_type, payload): + events.append(event_type) + + try: + answer = await AgentLoop(provider, registry, store, max_tool_output_chars=10_000).run( + spec=get_model_spec("work-light"), + messages=[{"role": "user", "content": "写脚本对比排序算法并生成报告"}], + identity=UserIdentity("u1", "", "", "user"), + raw_user_jwt="jwt", + chat_id="c1", + callback=callback, + ) + assert answer == "脚本运行和报告验证完成" + assert registry.exec_calls == 2 + assert events.count("completion.rejected") == 1 + recovery = provider.requests[4]["messages"][-1] + assert recovery["role"] == "user" + assert "SyntaxError" in recovery["content"] + finally: + await store.close() + + +async def test_script_delivery_requires_successful_execution(settings) -> None: + write = { + "choices": [ + { + "message": { + "role": "assistant", + "tool_calls": [tool_call("1", "write_file", {"path": "script.py", "content": "print('ok')"})], + }, + "finish_reason": "tool_calls", + } + ] + } + read = { + "choices": [ + { + "message": { + "role": "assistant", + "tool_calls": [tool_call("2", "read_file", {"path": "script.py"})], + }, + "finish_reason": "tool_calls", + } + ] + } + premature = {"choices": [{"message": {"role": "assistant", "content": "脚本完成"}, "finish_reason": "stop"}]} + execute = { + "choices": [ + { + "message": { + "role": "assistant", + "tool_calls": [tool_call("3", "exec", {"command": "python3 script.py"})], + }, + "finish_reason": "tool_calls", + } + ] + } + final = {"choices": [{"message": {"role": "assistant", "content": "脚本执行验证完成"}, "finish_reason": "stop"}]} + provider = ScriptedProvider([write, read, premature, execute, final]) + registry = RecordingRegistry() + store = RuntimeStore(settings.database_url) + await store.initialize() + + async def callback(event_type, payload): + return None + + try: + answer = await AgentLoop(provider, registry, store, max_tool_output_chars=10_000).run( + spec=get_model_spec("work-light"), + messages=[{"role": "user", "content": "写一个 Python 脚本"}], + identity=UserIdentity("u1", "", "", "user"), + raw_user_jwt="jwt", + chat_id="c1", + callback=callback, + ) + assert answer == "脚本执行验证完成" + assert [name for name, _ in registry.calls] == ["write_file", "read_file", "exec"] + assert "no command or process completed successfully" in provider.requests[3]["messages"][-1]["content"] + finally: + await store.close() + + +async def test_rejected_checkpoint_does_not_spin_without_tool_calls(settings) -> None: + stopped = {"choices": [{"message": {"role": "assistant", "content": "已完成"}, "finish_reason": "stop"}]} + provider = ScriptedProvider([stopped, stopped, stopped]) + store = RuntimeStore(settings.database_url) + await store.initialize() + events = [] + + async def callback(event_type, payload): + events.append(event_type) + + try: + answer = await AgentLoop( + provider, + RecordingRegistry(), + store, + max_tool_output_chars=10_000, + ).run( + spec=get_model_spec("work-light"), + messages=[{"role": "user", "content": "写一个报告"}], + identity=UserIdentity("u1", "", "", "user"), + raw_user_jwt="jwt", + chat_id="c1", + callback=callback, + ) + assert answer.startswith("未完成：") + assert len(provider.requests) == 3 + assert events.count("completion.rejected") == 2 + assert events.count("completion.unverified") == 1 finally: await store.close() diff --git a/tests/test_runtime_api.py b/tests/test_runtime_api.py index 39badbc..e9bc54e 100644 --- a/tests/test_runtime_api.py +++ b/tests/test_runtime_api.py @@ -136,6 +136,25 @@ def test_chat_stream_hides_provider_model_and_preserves_native_tool_calls(settin assert "data: [DONE]" in response.text +def test_openwebui_background_task_bypasses_work_loop(settings, identity_jwt) -> None: + provider = FakeModelProvider() + app = create_app(settings, store=RuntimeStore(settings.database_url), provider=provider, tools=FakeTools()) + with TestClient(app) as client: + response = client.post( + "/v1/chat/completions", + headers=headers(settings, identity_jwt, "background-task"), + json={ + "model": "work-medium", + "messages": [{"role": "user", "content": "Generate tags"}], + "metadata": {"task": "tags_generation"}, + }, + ) + assert response.status_code == 200 + assert response.json()["model"] == "work-medium" + assert provider.completed == [] + assert provider.forwarded[-1]["model"] == "ChatGPT-5.6:Terra" + + def test_work_stream_never_emits_delta_tool_calls(settings, identity_jwt) -> None: app = create_app( settings,