diff --git a/REFACTOR.md b/REFACTOR.md index ea5108ad..ba69991c 100644 --- a/REFACTOR.md +++ b/REFACTOR.md @@ -58,8 +58,11 @@ - [x] 将 Qwen-Image 修改为 Skill - [x] 现在输入区域对于不同 mention 的渲染的 ICON 和 human-message 里面的渲染的 ICON; - [x] 节点的颜色按照 label 的分类来 -- [ ] 添加推荐 Skill 的列表，点击可以直接安装，实际上是替代了填入搜索的名称，并触发拉取，这个似乎只需要在前端修改就可以了 +- [x] 添加推荐 Skill 的列表，点击可以直接安装，实际上是替代了填入搜索的名称，并触发拉取，这个似乎只需要在前端修改就可以了 +- [ ] 智能体调用的时候模型支持单独配置 +- [ ] 添加通用任务的子智能体 +- [ ] 添加深度研究智能体 # bug -- [ ] 知识库、知识图谱、评估基准的空状态是不一样的，需要统一 +- [x] 知识库、知识图谱、评估基准的空状态是不一样的，需要统一 diff --git a/backend/package/yuxi/services/agent_run_service.py b/backend/package/yuxi/services/agent_run_service.py index 51ffadb0..6148361b 100644 --- a/backend/package/yuxi/services/agent_run_service.py +++ b/backend/package/yuxi/services/agent_run_service.py @@ -13,6 +13,7 @@ from sqlalchemy import select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from yuxi.agents.buildin import agent_manager +from yuxi.models.providers.cache import model_cache from yuxi.repositories.agent_repository import AgentRepository from yuxi.repositories.agent_run_repository import TERMINAL_RUN_STATUSES, AgentRunRepository from yuxi.repositories.conversation_repository import ConversationRepository @@ -34,6 +35,30 @@ SSE_MAX_CONNECTION_MINUTES = int(os.getenv("RUN_SSE_MAX_CONNECTION_MINUTES", "30 SSE_POLL_INTERVAL_SECONDS = float(os.getenv("RUN_SSE_POLL_INTERVAL_SECONDS", "1.0")) +def _validate_model_spec(model_spec: str | None) -> str | None: + """校验对话级模型覆盖：未提供则返回 None；非法模型直接 422，不静默回退。""" + if not model_spec: + return None + info = model_cache.get_model_info(model_spec) + if not info or info.model_type != "chat": + raise HTTPException(status_code=422, detail=f"未找到可用聊天模型: '{model_spec}'") + return model_spec + + +def _resolve_effective_model_spec(model_spec: str | None, agent_item, agent_backend) -> str | None: + """解析本次 chat run 实际使用的模型：显式覆盖优先，否则快照智能体当前配置。""" + resolved_model_spec = _validate_model_spec(model_spec) + if resolved_model_spec: + return resolved_model_spec + + context = agent_backend.context_schema() + config_json = getattr(agent_item, "config_json", None) or {} + config_context = config_json.get("context") if isinstance(config_json, dict) else {} + if isinstance(config_context, dict): + context.update_from_dict(config_context) + return getattr(context, "model", None) + + def _build_run_response(run) -> dict: return { "run_id": run.id, @@ -65,6 +90,7 @@ async def create_agent_run_view( image_content: str | None, current_uid: str, db: AsyncSession, + model_spec: str | None = None, resume: object | None = None, parent_run_id: str | None = None, resume_request_id: str | None = None, @@ -91,13 +117,18 @@ async def create_agent_run_view( agent_item = await agent_repo.get_visible_by_slug(slug=agent_id, user=current_user) if not agent_item: raise HTTPException(status_code=404, detail="智能体不存在") - if not agent_manager.get_agent(agent_item.backend_id): + agent_backend = agent_manager.get_agent(agent_item.backend_id) + if not agent_backend: raise HTTPException(status_code=404, detail=f"智能体后端 {agent_item.backend_id} 不存在") run_type = "resume" if resume is not None else "chat" request_id = str(resume_request_id or (meta or {}).get("request_id") or uuid.uuid4()) config = {"thread_id": thread_id, "agent_id": agent_id} run_repo = AgentRunRepository(db) + # chat：快照本次实际模型；resume：沿用被恢复运行的原始模型，保证单次运行模型一致。 + resolved_model_spec = ( + _resolve_effective_model_spec(model_spec, agent_item, agent_backend) if run_type == "chat" else None + ) if run_type == "resume": if not parent_run_id: raise HTTPException(status_code=422, detail="parent_run_id 不能为空") @@ -106,6 +137,7 @@ async def create_agent_run_view( raise HTTPException(status_code=404, detail="被恢复的运行任务不存在") if parent_run.status != "interrupted": raise HTTPException(status_code=409, detail="只有 interrupted run 可以恢复") + resolved_model_spec = (parent_run.input_payload or {}).get("model_spec") if resume_request_id: existing_resume = await run_repo.get_resume_run(parent_run_id, resume_request_id) if existing_resume and existing_resume.uid == str(current_uid): @@ -125,6 +157,7 @@ async def create_agent_run_view( "run_type": run_type, "config": config or {}, "image_content": image_content, + "model_spec": resolved_model_spec, "agent_id": agent_id, "backend_id": agent_item.backend_id, "thread_id": thread_id, @@ -155,6 +188,7 @@ async def create_agent_run_view( "parent_run_id": parent_run_id, "resume": resume, "attachments": [], + "model_spec": resolved_model_spec, } if run_type == "resume": input_metadata["source"] = "ask_user_question_resume" diff --git a/backend/package/yuxi/services/chat_service.py b/backend/package/yuxi/services/chat_service.py index c6cd1a10..4ccf88e2 100644 --- a/backend/package/yuxi/services/chat_service.py +++ b/backend/package/yuxi/services/chat_service.py @@ -166,6 +166,13 @@ def _json_safe(value: Any) -> Any: return str(value) +def _apply_model_override(input_context: dict, meta: dict | None) -> None: + """对话级模型覆盖：meta.model_spec 优先于智能体配置的 model。值已在创建 run 时校验。""" + model_spec = (meta or {}).get("model_spec") + if model_spec: + input_context["model"] = model_spec + + def _stream_message_key(metadata: dict | None, namespace: list[str], thread_id: str | None) -> tuple[str, str]: if not isinstance(metadata, dict): return thread_id or "", "/".join(namespace) @@ -755,6 +762,7 @@ async def agent_chat( run_id=meta.get("run_id"), request_id=meta.get("request_id"), ) + _apply_model_override(input_context, meta) langfuse_run = _build_langfuse_run_context( current_user=current_user, thread_id=thread_id, @@ -961,6 +969,7 @@ async def stream_agent_chat( run_id=meta.get("run_id"), request_id=meta.get("request_id"), ) + _apply_model_override(input_context, meta) langfuse_run = _build_langfuse_run_context( current_user=current_user, thread_id=thread_id, @@ -1247,6 +1256,7 @@ async def stream_agent_resume( run_id=meta.get("run_id"), request_id=meta.get("request_id"), ) + _apply_model_override(input_context, meta) context = agent.context_schema() context.update(input_context) langfuse_run = _build_langfuse_run_context( diff --git a/backend/package/yuxi/services/run_worker.py b/backend/package/yuxi/services/run_worker.py index 3e48ff6c..8d966117 100644 --- a/backend/package/yuxi/services/run_worker.py +++ b/backend/package/yuxi/services/run_worker.py @@ -312,6 +312,7 @@ async def process_agent_run(ctx, run_id: str): "uid": user.uid, "has_image": bool(image_content), "attachment_file_ids": payload.get("attachment_file_ids") or [], + "model_spec": payload.get("model_spec"), } await mark_run_running(run_id) diff --git a/backend/server/routers/agent_router.py b/backend/server/routers/agent_router.py index 8e501eba..8614f409 100644 --- a/backend/server/routers/agent_router.py +++ b/backend/server/routers/agent_router.py @@ -57,6 +57,7 @@ class AgentRunCreate(BaseModel): thread_id: str = Field(..., description="会话线程 ID") meta: dict = Field(default_factory=dict, description="可选，请求追踪信息，例如 request_id") image_content: str | None = Field(None, description="可选，base64 图片内容") + model_spec: str | None = Field(None, description="可选，对话级模型覆盖，优先级高于智能体配置") resume: Any | None = Field(None, description="可选，恢复 interrupted run 的输入") parent_run_id: str | None = Field(None, description="可选，被恢复的 run ID") resume_request_id: str | None = Field(None, description="可选，resume 幂等键") @@ -258,6 +259,7 @@ async def create_agent_run( thread_id=payload.thread_id, meta=dict(payload.meta or {}), image_content=payload.image_content, + model_spec=payload.model_spec, current_uid=str(current_user.uid), db=db, resume=payload.resume, diff --git a/backend/test/unit/services/test_agent_run_service.py b/backend/test/unit/services/test_agent_run_service.py index e8c0a096..dbc90d4c 100644 --- a/backend/test/unit/services/test_agent_run_service.py +++ b/backend/test/unit/services/test_agent_run_service.py @@ -8,6 +8,20 @@ import pytest import yuxi.services.agent_run_service as agent_run_service +class _FakeContext: + def __init__(self): + self.model = "agent-default-model" + + def update_from_dict(self, data: dict): + for key, value in data.items(): + if hasattr(self, key): + setattr(self, key, value) + + +class _FakeBackend: + context_schema = _FakeContext + + @pytest.mark.asyncio async def test_stream_agent_run_events_emits_error_on_db_error(monkeypatch: pytest.MonkeyPatch): @asynccontextmanager @@ -140,6 +154,7 @@ async def test_create_agent_run_persists_input_before_enqueue(monkeypatch: pytes raise AssertionError("rollback should not be called") db = FakeDB() + captured = {} created_run = SimpleNamespace( id="", thread_id="thread-1", @@ -159,6 +174,7 @@ async def test_create_agent_run_persists_input_before_enqueue(monkeypatch: pytes async def create_run(self, **kwargs): assert kwargs["request_id"] == "req-1" assert kwargs["conversation_id"] == 1 + captured["input_payload"] = kwargs["input_payload"] created_run.id = kwargs["run_id"] return created_run @@ -194,7 +210,7 @@ async def test_create_agent_run_persists_input_before_enqueue(monkeypatch: pytes async def fake_get_arq_pool(): return Queue() - monkeypatch.setattr(agent_run_service.agent_manager, "get_agent", lambda backend_id: object()) + monkeypatch.setattr(agent_run_service.agent_manager, "get_agent", lambda backend_id: _FakeBackend()) monkeypatch.setattr(agent_run_service, "AgentRepository", AgentRepo) monkeypatch.setattr(agent_run_service, "ConversationRepository", ConvRepo) monkeypatch.setattr(agent_run_service, "AgentRunRepository", RunRepo) @@ -215,6 +231,8 @@ async def test_create_agent_run_persists_input_before_enqueue(monkeypatch: pytes assert result["request_id"] == "req-1" assert db.added[0].run_id == created_run.id assert db.added[0].request_id == "req-1" + assert captured["input_payload"]["model_spec"] == "agent-default-model" + assert db.added[0].extra_metadata["model_spec"] == "agent-default-model" @pytest.mark.asyncio @@ -265,7 +283,7 @@ async def test_create_resume_run_marks_input_message_source(monkeypatch: pytest. async def get_run_for_user(self, run_id: str, uid: str): assert run_id == "parent-run" assert uid == "user-1" - return SimpleNamespace(id=run_id, thread_id="thread-1", status="interrupted") + return SimpleNamespace(id=run_id, thread_id="thread-1", status="interrupted", input_payload={}) async def get_resume_run(self, parent_run_id: str, resume_request_id: str): assert parent_run_id == "parent-run" @@ -314,7 +332,7 @@ async def test_create_resume_run_marks_input_message_source(monkeypatch: pytest. async def fake_get_arq_pool(): return Queue() - monkeypatch.setattr(agent_run_service.agent_manager, "get_agent", lambda backend_id: object()) + monkeypatch.setattr(agent_run_service.agent_manager, "get_agent", lambda backend_id: _FakeBackend()) monkeypatch.setattr(agent_run_service, "AgentRepository", AgentRepo) monkeypatch.setattr(agent_run_service, "ConversationRepository", ConvRepo) monkeypatch.setattr(agent_run_service, "AgentRunRepository", RunRepo) @@ -336,3 +354,247 @@ async def test_create_resume_run_marks_input_message_source(monkeypatch: pytest. assert result["run_id"] == created_run.id assert db.added[0].message_type == "resume" assert db.added[0].extra_metadata["source"] == "ask_user_question_resume" + + +# ==================== 对话级模型覆盖 model_spec ==================== + + +def test_validate_model_spec_returns_none_when_empty(): + assert agent_run_service._validate_model_spec(None) is None + assert agent_run_service._validate_model_spec("") is None + + +def test_validate_model_spec_rejects_unknown_model(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr(agent_run_service.model_cache, "get_model_info", lambda spec: None) + with pytest.raises(agent_run_service.HTTPException) as exc: + agent_run_service._validate_model_spec("nope") + assert exc.value.status_code == 422 + + +def test_validate_model_spec_rejects_non_chat_model(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr( + agent_run_service.model_cache, + "get_model_info", + lambda spec: SimpleNamespace(model_type="embedding"), + ) + with pytest.raises(agent_run_service.HTTPException) as exc: + agent_run_service._validate_model_spec("embed-1") + assert exc.value.status_code == 422 + + +def test_validate_model_spec_accepts_chat_model(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr( + agent_run_service.model_cache, + "get_model_info", + lambda spec: SimpleNamespace(model_type="chat"), + ) + assert agent_run_service._validate_model_spec("gpt-x") == "gpt-x" + + +def _patch_common_run_repos( + monkeypatch: pytest.MonkeyPatch, + run_repo_cls, + *, + agent_config_json: dict | None = None, +): + class FakeResult: + def scalar_one_or_none(self): + return SimpleNamespace(uid="user-1", role="user") + + class FakeDB: + def __init__(self): + self.added = [] + self.committed = False + + async def execute(self, stmt): + del stmt + return FakeResult() + + def add(self, item): + self.added.append(item) + + async def flush(self): + for item in self.added: + if getattr(item, "id", None) is None: + item.id = 10 + + async def commit(self): + self.committed = True + + async def rollback(self): + raise AssertionError("rollback should not be called") + + class ConvRepo: + def __init__(self, db_session): + del db_session + + async def get_conversation_by_thread_id(self, thread_id: str): + del thread_id + return SimpleNamespace(id=1, uid="user-1", status="active", agent_id="default") + + class AgentRepo: + def __init__(self, db_session): + del db_session + + async def get_visible_by_slug(self, slug: str, user): + del user + return SimpleNamespace( + slug=slug, + backend_id="ChatbotAgent", + config_json=agent_config_json or {"context": {}}, + ) + + class Queue: + async def enqueue_job(self, job_name: str, run_id: str, _job_id: str): + del job_name, run_id, _job_id + + async def fake_get_arq_pool(): + return Queue() + + monkeypatch.setattr(agent_run_service.agent_manager, "get_agent", lambda backend_id: _FakeBackend()) + monkeypatch.setattr(agent_run_service, "AgentRepository", AgentRepo) + monkeypatch.setattr(agent_run_service, "ConversationRepository", ConvRepo) + monkeypatch.setattr(agent_run_service, "AgentRunRepository", run_repo_cls) + monkeypatch.setattr(agent_run_service, "get_arq_pool", fake_get_arq_pool) + return FakeDB() + + +@pytest.mark.asyncio +async def test_create_chat_run_persists_validated_model_spec(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr( + agent_run_service.model_cache, + "get_model_info", + lambda spec: SimpleNamespace(model_type="chat"), + ) + captured = {} + created_run = SimpleNamespace(id="", thread_id="thread-1", status="pending", request_id="req-1", uid="user-1") + + class RunRepo: + def __init__(self, db_session): + del db_session + + async def get_run_by_request_id(self, request_id: str): + del request_id + return None + + async def create_run(self, **kwargs): + captured["input_payload"] = kwargs["input_payload"] + created_run.id = kwargs["run_id"] + return created_run + + async def set_input_message(self, run_id: str, message_id: int): + del run_id, message_id + return created_run + + db = _patch_common_run_repos(monkeypatch, RunRepo) + + await agent_run_service.create_agent_run_view( + query="hello", + agent_id="default", + thread_id="thread-1", + meta={"request_id": "req-1"}, + image_content=None, + current_uid="user-1", + db=db, + model_spec="claude-x", + ) + + assert captured["input_payload"]["model_spec"] == "claude-x" + + +@pytest.mark.asyncio +async def test_create_chat_run_snapshots_agent_configured_model_spec(monkeypatch: pytest.MonkeyPatch): + captured = {} + created_run = SimpleNamespace(id="", thread_id="thread-1", status="pending", request_id="req-1", uid="user-1") + + class RunRepo: + def __init__(self, db_session): + del db_session + + async def get_run_by_request_id(self, request_id: str): + del request_id + return None + + async def create_run(self, **kwargs): + captured["input_payload"] = kwargs["input_payload"] + created_run.id = kwargs["run_id"] + return created_run + + async def set_input_message(self, run_id: str, message_id: int): + del run_id, message_id + return created_run + + db = _patch_common_run_repos( + monkeypatch, + RunRepo, + agent_config_json={"context": {"model": "agent-config-model"}}, + ) + + await agent_run_service.create_agent_run_view( + query="hello", + agent_id="default", + thread_id="thread-1", + meta={"request_id": "req-1"}, + image_content=None, + current_uid="user-1", + db=db, + model_spec=None, + ) + + assert captured["input_payload"]["model_spec"] == "agent-config-model" + assert db.added[0].extra_metadata["model_spec"] == "agent-config-model" + + +@pytest.mark.asyncio +async def test_create_resume_run_inherits_parent_model_spec(monkeypatch: pytest.MonkeyPatch): + # 即使 resume 入参传了别的模型，也必须沿用父运行的模型 + captured = {} + created_run = SimpleNamespace(id="", thread_id="thread-1", status="pending", request_id="resume-req", uid="user-1") + + class RunRepo: + def __init__(self, db_session): + del db_session + + async def get_run_for_user(self, run_id: str, uid: str): + del uid + return SimpleNamespace( + id=run_id, + thread_id="thread-1", + status="interrupted", + input_payload={"model_spec": "parent-model"}, + ) + + async def get_resume_run(self, parent_run_id: str, resume_request_id: str): + del parent_run_id, resume_request_id + return None + + async def get_run_by_request_id(self, request_id: str): + del request_id + return None + + async def create_run(self, **kwargs): + captured["input_payload"] = kwargs["input_payload"] + created_run.id = kwargs["run_id"] + return created_run + + async def set_input_message(self, run_id: str, message_id: int): + del run_id, message_id + return created_run + + db = _patch_common_run_repos(monkeypatch, RunRepo) + + await agent_run_service.create_agent_run_view( + query=None, + agent_id="default", + thread_id="thread-1", + meta={"request_id": "resume-req"}, + image_content=None, + current_uid="user-1", + db=db, + model_spec="ignored-model", + resume={"language": "python"}, + parent_run_id="parent-run", + resume_request_id="resume-req", + ) + + assert captured["input_payload"]["model_spec"] == "parent-model" diff --git a/backend/test/unit/services/test_chat_service_model_override.py b/backend/test/unit/services/test_chat_service_model_override.py new file mode 100644 index 00000000..e5b5d59b --- /dev/null +++ b/backend/test/unit/services/test_chat_service_model_override.py @@ -0,0 +1,16 @@ +from __future__ import annotations + +from yuxi.services import chat_service as svc + + +def test_apply_model_override_sets_model_from_meta(): + input_context = {"model": "agent-default"} + svc._apply_model_override(input_context, {"model_spec": "user-pick"}) + assert input_context["model"] == "user-pick" + + +def test_apply_model_override_noop_without_model_spec(): + input_context = {"model": "agent-default"} + svc._apply_model_override(input_context, {"request_id": "r1"}) + svc._apply_model_override(input_context, None) + assert input_context["model"] == "agent-default" diff --git a/docs/develop-guides/roadmap.md b/docs/develop-guides/roadmap.md index 1b89ee15..320a0a7a 100644 --- a/docs/develop-guides/roadmap.md +++ b/docs/develop-guides/roadmap.md @@ -72,6 +72,7 @@ - 工作区文件上传支持多选：`/workspace/upload` 与 Viewer 工作区上传统一使用 `files` 多文件字段，一次最多上传 50 个文件，批量上传失败时清理本次已写入文件。 - 聊天附件新增 MinIO tmp 临时上传、可选 PDF/图片解析、确认后加入线程附件的流程；前端改为弹窗内上传、解析与确认。 - 标准化 Agent run/SSE 执行链路：run 创建时持久化输入消息并提交后入队，worker 统一写入 Redis Stream envelope，SSE 输出 `event/data/id`、心跳注释、`Last-Event-ID` 回放和终止 `end` 事件；前端强制使用 run API 并支持 ask_user_question 中断后以 resume run 恢复；事件 envelope 构造收敛到统一 helper，前端优先使用 envelope 一级 `thread_id` 路由。 +- 新增对话级模型覆盖：发起 run 时可携带 `model_spec` 作为运行时模型覆盖，优先级高于智能体配置，无需智能体写权限即可在对话中切换模型；模型校验在创建 run 时完成（非法直接 422，不静默回退），并随 `agent_runs.input_payload` 持久化留痕，resume 沿用被恢复运行的原始模型；前端对话输入区接入模型选择器，按线程记忆选择、默认回退到智能体配置模型，未显式切换时下发 `null` 由后端使用智能体配置模型。输入消息 `extra_metadata` 记录所用 `model_spec`，重开历史会话时按最近一条用户消息还原模型选择。 - 收敛后端模块边界：文档解析从 `plugins.parser` 移动到 `knowledge.parser`，内容审查从 `plugins.guard` 移动到 `services.guard`。 - 收敛文件服务边界：文件预览判断抽为独立服务，Viewer 文件系统的 workspace 分支复用用户 workspace 服务，线程运行时上下文解析从泛化 `filesystem_service` 拆出为 agent runtime helper。 - 升级 DeepAgents 到 0.6.7 并适配新版文件系统协议：SubAgentMiddleware 改为显式 subagent spec，Skills prompt 补齐新版占位符；sandbox/skills backend 复用新版 `ReadResult`、`GlobResult`、`GrepResult` 等协议类型，文件权限在 backend 层明确区分 skills、uploads、outputs 与 workspace，保留最小 `CustomCompositeBackend` 以避免非 route glob 误扫其他 route；Agent 上下文压缩改为复用 DeepAgents SummarizationMiddleware，历史摘要与大工具结果统一 offload 到 outputs。 diff --git a/web/src/apis/agent_api.js b/web/src/apis/agent_api.js index 99c952d0..925287d4 100644 --- a/web/src/apis/agent_api.js +++ b/web/src/apis/agent_api.js @@ -105,6 +105,7 @@ export const agentApi = { thread_id: data.thread_id, meta: data.meta || {}, image_content: data.image_content || null, + model_spec: data.model_spec || null, resume: data.resume ?? null, parent_run_id: data.parent_run_id || null, resume_request_id: data.resume_request_id || null diff --git a/web/src/components/AgentChatComponent.vue b/web/src/components/AgentChatComponent.vue index 0dfd106f..190090f5 100644 --- a/web/src/components/AgentChatComponent.vue +++ b/web/src/components/AgentChatComponent.vue @@ -133,6 +133,15 @@ @remove-attachment="handleAttachmentRemove" >