[codex] Add multi-agent orchestration, tracing, and tool policies

README.md

@@ -33,6 +33,9 @@ SearchAgent 是一个本地优先的 AI 深度研究工作台。它用类似 Cod
 - OpenAI、Anthropic、DeepSeek、Xiaomi MiMo、Alibaba Qwen 和自定义服务预设
 - 本地 stdio 与远程 HTTP MCP 服务配置、测试和编辑
 - API Key 与 MCP 环境变量本地加密存储
+- 规划、检索、核验、写作四类 Agent 任务编排，检索任务最多三路协作
+- 本地 Trace、工具调用审计、Agent 级工具策略和运行中授权确认
+- 本地评测数据集、确定性指标及可选独立 LLM 裁判配置
 - 中英文界面、浏览器语言检测和语言偏好持久化
 
 ## 技术栈
@@ -123,6 +126,8 @@ SearchAgent 会识别名为 `bocha` 的配置，并使用保存的密钥调用
 └── reports/         # 生成的报告导出文件
 ```
 
+Trace、Agent 任务、工具策略、授权记录和评测结果也保存在 `searchagent.db`。权限策略按 Agent 角色和工具名匹配，未知工具默认暂停等待本次任务确认；批准不会写入持久化允许规则。首期是应用层隔离，不提供 MCP 进程的操作系统级沙箱。
+
 可以用 `SEARCHAGENT_HOME` 指定其他目录：
 
 ```powershell

backend/app/api/observability.py

@@ -0,0 +1,152 @@
+import asyncio
+import time
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+
+from app.api.deps import get_db
+from app.db.models import Conversation, EvaluationCase, EvaluationDataset, EvaluationRun, EvaluationScore, LLMProfile, ResearchProject
+from app.engine.runner import resume_research, start_research
+from app.llm.factory import build_chat_model_from_profile
+from app.schemas.observability import EvaluationCasePayload, EvaluationDatasetPayload, EvaluationRunPayload, ToolApprovalPayload, ToolPolicyPayload
+from app.services.tracing_service import list_trace
+from app.tools import policy
+
+router = APIRouter(tags=["observability"])
+
+
+@router.post("/research/{thread_id}/tool-approval")
+async def approve_tool_call(thread_id: str, payload: ToolApprovalPayload, session: Session = Depends(get_db)):
+    from app.engine.runner import resume_research
+
+    return await resume_research(
+        session,
+        thread_id=thread_id,
+        profile_id=payload.profile_id,
+        decision={"kind": "tool_approval", "approved": payload.approved, "agent_role": payload.agent_role, "tool_name": payload.tool_name, "args_fingerprint": payload.args_fingerprint},
+    )
+
+
+@router.get("/tool-policies")
+def list_tool_policies(session: Session = Depends(get_db)):
+    return policy.list_policies(session)
+
+
+@router.post("/tool-policies")
+def create_tool_policy(payload: ToolPolicyPayload, session: Session = Depends(get_db)):
+    return policy.create_policy(session, **payload.model_dump())
+
+
+@router.put("/tool-policies/{policy_id}")
+def update_tool_policy(policy_id: int, payload: ToolPolicyPayload, session: Session = Depends(get_db)):
+    result = policy.update_policy(session, policy_id, **payload.model_dump())
+    if result is None:
+        raise HTTPException(status_code=404, detail="Tool policy not found")
+    return result
+
+
+@router.delete("/tool-policies/{policy_id}", status_code=204)
+def delete_tool_policy(policy_id: int, session: Session = Depends(get_db)):
+    if not policy.delete_policy(session, policy_id):
+        raise HTTPException(status_code=404, detail="Tool policy not found")
+
+
+@router.get("/research/projects/{project_id}/traces")
+def get_project_traces(project_id: int, session: Session = Depends(get_db)):
+    if session.get(ResearchProject, project_id) is None:
+        raise HTTPException(status_code=404, detail="Research project not found")
+    return list_trace(session, project_id)
+
+
+@router.post("/evaluation/datasets")
+def create_dataset(payload: EvaluationDatasetPayload, session: Session = Depends(get_db)):
+    if payload.judge_profile_id and session.get(LLMProfile, payload.judge_profile_id) is None:
+        raise HTTPException(status_code=404, detail="Judge profile not found")
+    item = EvaluationDataset(**payload.model_dump())
+    session.add(item)
+    session.commit()
+    return _dataset(item, session)
+
+
+@router.get("/evaluation/datasets")
+def list_datasets(session: Session = Depends(get_db)):
+    return [_dataset(item, session) for item in session.query(EvaluationDataset).order_by(EvaluationDataset.id.desc())]
+
+
+@router.post("/evaluation/datasets/{dataset_id}/cases")
+def create_case(dataset_id: int, payload: EvaluationCasePayload, session: Session = Depends(get_db)):
+    if session.get(EvaluationDataset, dataset_id) is None:
+        raise HTTPException(status_code=404, detail="Evaluation dataset not found")
+    item = EvaluationCase(dataset_id=dataset_id, input_text=payload.input_text, expected_json=payload.expected)
+    session.add(item)
+    session.commit()
+    return {"id": item.id, "dataset_id": item.dataset_id, "input_text": item.input_text, "expected": item.expected_json}
+
+
+@router.post("/evaluation/datasets/{dataset_id}/runs")
+def run_dataset(dataset_id: int, payload: EvaluationRunPayload, session: Session = Depends(get_db)):
+    dataset = session.get(EvaluationDataset, dataset_id)
+    if dataset is None:
+        raise HTTPException(status_code=404, detail="Evaluation dataset not found")
+    if session.get(LLMProfile, payload.profile_id) is None:
+        raise HTTPException(status_code=404, detail="Research profile not found")
+    judge_id = payload.judge_profile_id if payload.judge_profile_id is not None else dataset.judge_profile_id
+    if judge_id is not None and session.get(LLMProfile, judge_id) is None:
+        raise HTTPException(status_code=404, detail="Judge profile not found")
+    started = time.perf_counter()
+    cases = session.query(EvaluationCase).filter(EvaluationCase.dataset_id == dataset_id).all()
+    run = EvaluationRun(dataset_id=dataset_id, profile_id=payload.profile_id, judge_profile_id=judge_id, status="completed")
+    session.add(run)
+    session.flush()
+    for case in cases:
+        metrics = {"input_nonempty": bool(case.input_text.strip()), "expected_present": case.expected_json is not None, "source_count": 0, "citation_complete": False, "permission_violations": 0, "interrupted": False}
+        status = "failed"
+        if metrics["input_nonempty"]:
+            conversation = Conversation(title=f"Evaluation {dataset.name} #{case.id}")
+            session.add(conversation)
+            session.commit()
+            try:
+                result = asyncio.run(start_research(session, conversation_id=conversation.id, profile_id=payload.profile_id, user_message=case.input_text))
+                metrics["interrupted"] = bool(result.get("interrupted"))
+                if result.get("interrupted") and (result.get("interrupt_payload") or {}).get("kind") in {"plan_ready", "plan_approval"}:
+                    version = (result.get("state") or {}).get("plan_version")
+                    resumed = asyncio.run(resume_research(session, thread_id=result["thread_id"], profile_id=payload.profile_id, decision={"kind": "execute_plan", "plan_version": version}))
+                    result = resumed
+                state = result.get("state") or {}
+                findings = state.get("verified_findings") or state.get("findings") or []
+                metrics["source_count"] = len(findings)
+                metrics["citation_complete"] = bool(state.get("report_md") and "[^" in str(state.get("report_md")))
+                status = "passed" if state.get("report_id") else "blocked"
+                if judge_id is not None and state.get("report_md"):
+                    try:
+                        judge = build_chat_model_from_profile(session, judge_id)
+                        verdict = judge.invoke(
+                            "Score this research report from 0 to 1 for whether it addresses the expected criteria. "
+                            "Return only a number.\n"
+                            f"Expected: {case.expected_json or {}}\nReport: {str(state['report_md'])[:6000]}"
+                        )
+                        metrics["judge_score"] = float(str(getattr(verdict, "content", verdict)).strip())
+                    except Exception as exc:
+                        metrics["judge_error"] = str(exc)[:300]
+            except Exception as exc:
+                metrics["error"] = str(exc)[:500]
+                status = "failed"
+        session.add(EvaluationScore(run_id=run.id, case_id=case.id, status=status, metrics_json=metrics))
+    run.metrics_json = {"case_count": len(cases), "completed": len(cases), "latency_ms": round((time.perf_counter() - started) * 1000, 2), "judge_profile_id": judge_id}
+    session.commit()
+    return _run(run, session)
+
+
+@router.get("/evaluation/runs")
+def list_runs(session: Session = Depends(get_db)):
+    return [_run(item, session) for item in session.query(EvaluationRun).order_by(EvaluationRun.id.desc())]
+
+
+def _dataset(item: EvaluationDataset, session: Session) -> dict:
+    cases = session.query(EvaluationCase).filter(EvaluationCase.dataset_id == item.id).all()
+    return {"id": item.id, "name": item.name, "version": item.version, "description": item.description, "judge_profile_id": item.judge_profile_id, "case_count": len(cases), "created_at": item.created_at.isoformat()}
+
+
+def _run(item: EvaluationRun, session: Session) -> dict:
+    scores = session.query(EvaluationScore).filter(EvaluationScore.run_id == item.id).all()
+    return {"id": item.id, "dataset_id": item.dataset_id, "profile_id": item.profile_id, "judge_profile_id": item.judge_profile_id, "status": item.status, "metrics": item.metrics_json, "scores": [{"case_id": score.case_id, "status": score.status, "metrics": score.metrics_json} for score in scores], "created_at": item.created_at.isoformat()}

backend/app/api/research.py

@@ -2,7 +2,7 @@
 from sqlalchemy.orm import Session
 
 from app.api.deps import get_db
-from app.db.models import Conversation, LLMProfile, ResearchProject, Source, Step
+from app.db.models import AgentTask, Conversation, LLMProfile, ResearchProject, Source, Step, ToolApproval, ToolCall
 from app.engine import runner
 from app.schemas.research import (
     ResearchResumeRequest,
@@ -116,6 +116,22 @@ def project_execution_detail(
             }
             for report in sorted(project.reports, key=lambda item: (item.version, item.id))
         ],
+        "agent_tasks": [
+            {"id": task.id, "role": task.role, "title": task.title, "status": task.status,
+             "input": task.input_json, "output": task.output_json}
+            for task in session.query(AgentTask).filter(AgentTask.project_id == project.id).order_by(AgentTask.id)
+        ],
+        "tool_calls": [
+            {"id": call.id, "task_id": call.task_id, "agent_role": call.agent_role,
+             "tool_name": call.tool_name, "status": call.status, "result_summary": call.result_summary,
+             "error": call.error}
+            for call in session.query(ToolCall).filter(ToolCall.project_id == project.id).order_by(ToolCall.id)
+        ],
+        "approvals": [
+            {"id": approval.id, "agent_role": approval.agent_role, "tool_name": approval.tool_name,
+             "decision": approval.decision, "args_fingerprint": approval.args_fingerprint}
+            for approval in session.query(ToolApproval).filter(ToolApproval.project_id == project.id).order_by(ToolApproval.id)
+        ],
     }
 
 

backend/app/api/router.py

@@ -1,6 +1,6 @@
 from fastapi import APIRouter
 
-from app.api import conversations, events, health, mcp, reports, research, settings
+from app.api import conversations, events, health, mcp, observability, reports, research, settings
 
 api_router = APIRouter()
 api_router.include_router(health.router)
@@ -10,3 +10,4 @@
 api_router.include_router(mcp.router)
 api_router.include_router(reports.router)
 api_router.include_router(events.router)
+api_router.include_router(observability.router)

backend/app/db/models.py

@@ -189,3 +189,134 @@ class AppPreference(Base):
 
     key: Mapped[str] = mapped_column(primary_key=True)
     value_json: Mapped[Optional[dict]] = mapped_column(JSON, default=None)
+
+
+class AgentTask(Base):
+    __tablename__ = "agent_tasks"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    project_id: Mapped[int] = mapped_column(ForeignKey("research_projects.id", ondelete="CASCADE"))
+    trace_id: Mapped[str] = mapped_column(index=True)
+    role: Mapped[str] = mapped_column()
+    title: Mapped[str] = mapped_column()
+    status: Mapped[str] = mapped_column(default="pending")
+    input_json: Mapped[Optional[dict]] = mapped_column(JSON, default=None)
+    output_json: Mapped[Optional[dict]] = mapped_column(JSON, default=None)
+    created_at: Mapped[dt.datetime] = mapped_column(default=_now)
+    started_at: Mapped[Optional[dt.datetime]] = mapped_column(default=None)
+    completed_at: Mapped[Optional[dt.datetime]] = mapped_column(default=None)
+
+
+class ToolPolicy(Base):
+    __tablename__ = "tool_policies"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    version: Mapped[int] = mapped_column(default=1)
+    agent_role: Mapped[str] = mapped_column(index=True)
+    tool_name: Mapped[str] = mapped_column(index=True)
+    allowed_domains_json: Mapped[Optional[list]] = mapped_column(JSON, default=None)
+    require_approval: Mapped[bool] = mapped_column(Boolean, default=False)
+    enabled: Mapped[bool] = mapped_column(Boolean, default=True)
+    created_at: Mapped[dt.datetime] = mapped_column(default=_now)
+
+
+class ToolApproval(Base):
+    __tablename__ = "tool_approvals"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    project_id: Mapped[int] = mapped_column(ForeignKey("research_projects.id", ondelete="CASCADE"))
+    trace_id: Mapped[str] = mapped_column(index=True)
+    agent_role: Mapped[str] = mapped_column()
+    tool_name: Mapped[str] = mapped_column()
+    args_fingerprint: Mapped[str] = mapped_column(index=True)
+    decision: Mapped[str] = mapped_column(default="approved")
+    created_at: Mapped[dt.datetime] = mapped_column(default=_now)
+
+
+class ToolCall(Base):
+    __tablename__ = "tool_calls"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    project_id: Mapped[int] = mapped_column(ForeignKey("research_projects.id", ondelete="CASCADE"))
+    trace_id: Mapped[str] = mapped_column(index=True)
+    task_id: Mapped[Optional[int]] = mapped_column(ForeignKey("agent_tasks.id", ondelete="SET NULL"), default=None)
+    agent_role: Mapped[str] = mapped_column()
+    tool_name: Mapped[str] = mapped_column()
+    args_json: Mapped[Optional[dict]] = mapped_column(JSON, default=None)
+    status: Mapped[str] = mapped_column(default="started")
+    result_summary: Mapped[Optional[str]] = mapped_column(Text, default=None)
+    error: Mapped[Optional[str]] = mapped_column(Text, default=None)
+    created_at: Mapped[dt.datetime] = mapped_column(default=_now)
+    completed_at: Mapped[Optional[dt.datetime]] = mapped_column(default=None)
+
+
+class TraceRun(Base):
+    __tablename__ = "trace_runs"
+
+    id: Mapped[str] = mapped_column(primary_key=True)
+    project_id: Mapped[int] = mapped_column(ForeignKey("research_projects.id", ondelete="CASCADE"), index=True)
+    root_name: Mapped[str] = mapped_column(default="research")
+    status: Mapped[str] = mapped_column(default="running")
+    created_at: Mapped[dt.datetime] = mapped_column(default=_now)
+    completed_at: Mapped[Optional[dt.datetime]] = mapped_column(default=None)
+
+
+class TraceSpan(Base):
+    __tablename__ = "trace_spans"
+
+    id: Mapped[str] = mapped_column(primary_key=True)
+    trace_id: Mapped[str] = mapped_column(ForeignKey("trace_runs.id", ondelete="CASCADE"), index=True)
+    parent_id: Mapped[Optional[str]] = mapped_column(ForeignKey("trace_spans.id", ondelete="SET NULL"), default=None)
+    name: Mapped[str] = mapped_column()
+    kind: Mapped[str] = mapped_column(default="internal")
+    status: Mapped[str] = mapped_column(default="running")
+    attributes_json: Mapped[Optional[dict]] = mapped_column(JSON, default=None)
+    input_summary: Mapped[Optional[str]] = mapped_column(Text, default=None)
+    output_summary: Mapped[Optional[str]] = mapped_column(Text, default=None)
+    error: Mapped[Optional[str]] = mapped_column(Text, default=None)
+    started_at: Mapped[dt.datetime] = mapped_column(default=_now)
+    ended_at: Mapped[Optional[dt.datetime]] = mapped_column(default=None)
+
+
+class EvaluationDataset(Base):
+    __tablename__ = "evaluation_datasets"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(unique=True)
+    version: Mapped[int] = mapped_column(default=1)
+    description: Mapped[str] = mapped_column(Text, default="")
+    judge_profile_id: Mapped[Optional[int]] = mapped_column(ForeignKey("llm_profiles.id", ondelete="SET NULL"), default=None)
+    created_at: Mapped[dt.datetime] = mapped_column(default=_now)
+
+
+class EvaluationCase(Base):
+    __tablename__ = "evaluation_cases"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    dataset_id: Mapped[int] = mapped_column(ForeignKey("evaluation_datasets.id", ondelete="CASCADE"))
+    input_text: Mapped[str] = mapped_column(Text)
+    expected_json: Mapped[Optional[dict]] = mapped_column(JSON, default=None)
+    created_at: Mapped[dt.datetime] = mapped_column(default=_now)
+
+
+class EvaluationRun(Base):
+    __tablename__ = "evaluation_runs"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    dataset_id: Mapped[int] = mapped_column(ForeignKey("evaluation_datasets.id", ondelete="CASCADE"))
+    profile_id: Mapped[int] = mapped_column(ForeignKey("llm_profiles.id", ondelete="RESTRICT"))
+    judge_profile_id: Mapped[Optional[int]] = mapped_column(ForeignKey("llm_profiles.id", ondelete="SET NULL"), default=None)
+    status: Mapped[str] = mapped_column(default="completed")
+    metrics_json: Mapped[Optional[dict]] = mapped_column(JSON, default=None)
+    created_at: Mapped[dt.datetime] = mapped_column(default=_now)
+
+
+class EvaluationScore(Base):
+    __tablename__ = "evaluation_scores"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    run_id: Mapped[int] = mapped_column(ForeignKey("evaluation_runs.id", ondelete="CASCADE"))
+    case_id: Mapped[int] = mapped_column(ForeignKey("evaluation_cases.id", ondelete="CASCADE"))
+    status: Mapped[str] = mapped_column()
+    metrics_json: Mapped[dict] = mapped_column(JSON, default=dict)
+    created_at: Mapped[dt.datetime] = mapped_column(default=_now)