Research Agent 实战：从 RAG 到自主研究

unzip research.zip -d research-demo
cd research-demo
python3.11 -m venv .venv
. .venv/bin/activate
python -m pip install --requirement requirements.txt
python app.py
python -m unittest -v test_app

mode=fixture (本地固定响应；无联网检索)
corpus=本地合成语料：虚构的南岚团队规程，不代表现实政策
plan.questions=2
tool.search=2 tool.fetch=2 notes=2
check revision=0 passed=false issues=unknown_citation:N999
check revision=1 passed=true issues=none
status=complete
citations=N1:S1,N2:S2
workspace=temporary (已清理；用 --workspace 显式保留)

research/
├── app.py                 # 参数、模型创建、结果保存、关闭客户端
├── pipeline.py            # 规划、研究工具循环、整理、检查和修订
├── sources.py             # 真实本地语料搜索与读取
├── fixture_model.py       # 明确标记的固定决策策略
├── fixtures/               # 合成语料清单与两份文本
├── test_app.py             # 包括成功、失败与超时的测试
├── agent_runtime.py        # ZIP 注入的共享异步运行层
└── requirements.txt        # ZIP 生成的固定版本依赖

{
  "label": "我的产品资料",
  "documents": [
    {"id": "Policy1", "title": "产品记录规则", "file": "policy.txt"}
  ]
}

python app.py --corpus ./my-corpus/corpus.json \
  --task '整理这份资料对研究记录的要求。' \
  --workspace ./research-output

"""Run the offline corpus demonstration, or an explicitly configured live model."""
from __future__ import annotations

import argparse
import asyncio
from dataclasses import asdict
import json
import math
from pathlib import Path
import sys
import tempfile
import uuid

# This example creates research data only in a temporary/explicit workspace.
sys.dont_write_bytecode = True

from agent_runtime import atomic_write_text, confined_path, model_from_env

if __package__:
    from .fixture_model import CorpusFixtureModel
    from .pipeline import Limits, render_report, run_research
    from .sources import CorpusProvider
else:
    from fixture_model import CorpusFixtureModel
    from pipeline import Limits, render_report, run_research
    from sources import CorpusProvider

FIXTURE_CORPUS = Path(__file__).parent / "fixtures" / "corpus.json"
DEFAULT_TASK = "整理南岚团队两份合成规程中的研究记录与发布复核要求。"


def positive_seconds(value: str) -> float:
    number = float(value)
    if not math.isfinite(number) or number <= 0:
        raise argparse.ArgumentTypeError("time must be finite and positive")
    return number


def parser():
    result = argparse.ArgumentParser(description=__doc__)
    result.add_argument("--task", default=DEFAULT_TASK)
    result.add_argument("--corpus", type=Path, help="explicit local corpus manifest (UTF-8 JSON)")
    result.add_argument("--live", action="store_true", help="use a live model; source tools remain local-only")
    result.add_argument("--workspace", type=Path, help="persist a new run subdirectory here")
    result.add_argument("--timeout", type=positive_seconds, default=60.0)
    result.add_argument("--request-timeout", type=positive_seconds, default=10.0)
    result.add_argument("--max-revisions", type=int, choices=range(6), default=2)
    return result


def save_run(result, workspace: Path):
    workspace.mkdir(parents=True, exist_ok=True)
    run_dir = confined_path(workspace, "run-" + uuid.uuid4().hex)
    run_dir.mkdir()
    for name, value in (("result", asdict(result)), ("plan", result.plan), ("sources", result.sources),
                        ("notes", result.notes), ("drafts", result.drafts), ("checks", result.checks)):
        atomic_write_text(confined_path(run_dir, name + ".json"),
                          json.dumps(value, ensure_ascii=False, indent=2, allow_nan=False) + "\n")
    atomic_write_text(confined_path(run_dir, "report.md"), render_report(result))
    return run_dir


async def async_main(args) -> int:
    model = None
    try:
        if args.live and args.corpus is None:
            raise ValueError("--live requires --corpus; only explicitly supplied local files are researched")
        provider = CorpusProvider(args.corpus or FIXTURE_CORPUS, request_timeout=args.request_timeout)
        limits = Limits(total_timeout=args.timeout, request_timeout=args.request_timeout,
                        max_revisions=args.max_revisions)
        model = model_from_env() if args.live else CorpusFixtureModel()
        mode = "live" if args.live else "fixture"
        print(f"mode={mode} (" + ("联网模型；资料工具只读本地文件" if args.live
                                   else "本地固定响应；无联网检索") + ")")
        print("corpus=" + provider.label)
        result = await run_research(args.task, model, provider, limits=limits, mode=mode)
        print(f'plan.questions={len(result.plan.get("questions", []))}')
        searches = sum(e["event"] == "search" for e in result.events)
        fetches = sum(e["event"] == "fetch" for e in result.events)
        print(f"tool.search={searches} tool.fetch={fetches} notes={len(result.notes)}")
        for check in result.checks:
            issues = ",".join(check["mechanical_issues"]) or "none"
            print(f'check revision={check["revision"]} passed={str(check["passed"]).lower()} issues={issues}')
        print("status=" + result.status)
        if result.report:
            used = {n for c in result.report["claims"] for n in c["note_ids"]}
            print("citations=" + ",".join(n["id"] + ":" + n["source_id"]
                                         for n in result.notes if n["id"] in used))
        for error in result.errors:
            print("error=" + error)
        if args.workspace is not None:
            location = save_run(result, args.workspace)
            print("workspace=" + str(location))
        else:
            with tempfile.TemporaryDirectory(prefix="research-agent-") as temporary:
                save_run(result, Path(temporary))
            print("workspace=temporary (已清理；用 --workspace 显式保留)")
        return 0 if result.status == "complete" else (124 if result.status == "timeout" else 1)
    except ValueError as error:
        print("configuration_error=" + str(error), file=sys.stderr)
        return 2
    except Exception as error:
        print("run_error=" + type(error).__name__, file=sys.stderr)
        return 1
    finally:
        if args.live and model is not None:
            await model.aclose()


def main(argv=None) -> int:
    args = parser().parse_args(argv)
    try:
        return asyncio.run(async_main(args))
    except KeyboardInterrupt:
        print("cancelled; work is incomplete", file=sys.stderr)
        return 130


if __name__ == "__main__":
    raise SystemExit(main())

python app.py --live --corpus ./my-corpus/corpus.json \
  --task '比较这些资料对记录与复核的要求，指出未覆盖的问题。' \
  --timeout 60 --request-timeout 10 --max-revisions 2 \
  --workspace ./research-output