ReAct：看一次结果，再决定下一步

mkdir "F:\学习\l3-react" -Force
cd "F:\学习\l3-react"

mkdir "F:\学习\l3-react" -Force
cd "F:\学习\l3-react"

"""通过两份固定资料观察动作与结果交替。"""
from __future__ import annotations

import argparse
import json


DOCUMENTS = {
    "primary": {"refund": "退款申请应在购买后 7 天内提交。"},
    "secondary": {"refund": "退款窗口为 7 天；超过窗口需人工审核。"},
}


# 场景名决定这次查询返回资料、空结果还是错误。
def search(source: str, query: str, scenario: str) -> str | None:
    if scenario == "tool-failure":
        raise RuntimeError("fixture_tool_unavailable")
    if scenario == "missing":
        return None
    if scenario == "found-second" and source == "primary":
        return None
    return DOCUMENTS[source].get(query)


# 每次动作后保存观察，找到资料时立即形成回答。
def run(scenario: str, max_actions: int = 2) -> dict:
    if scenario not in {"found-first", "found-second", "missing", "tool-failure", "forbidden-proposal"} or max_actions < 1:
        return {"status": "rejected", "reason": "invalid_input", "events": []}
    events: list[dict] = []
    for source in ("primary", "secondary"):
        if len([item for item in events if item["type"] == "action"]) >= max_actions:
            events.append({"type": "stop", "reason": "max_actions"})
            return {"status": "budget_exhausted", "answer": None, "events": events}
        proposed_tool = "delete_all" if scenario == "forbidden-proposal" else "search"
        events.append({"type": "scripted_proposal", "tool": proposed_tool, "source": source})
        allowed = proposed_tool == "search" and source in DOCUMENTS
        events.append({"type": "program_check", "allowed": allowed, "reason": "allowlisted_source_and_budget" if allowed else "tool_not_allowlisted"})
        if not allowed:
            events.append({"type": "stop", "reason": "forbidden_proposal"})
            return {"status": "rejected", "answer": None, "events": events}
        events.append({"type": "action", "tool": proposed_tool, "source": source, "query": "refund"})
        try:
            result = search(source, "refund", scenario)
        except RuntimeError as error:
            events.append({"type": "observation", "source": source, "error": str(error)})
            continue
        events.append({"type": "observation", "source": source, "text": result})
        if result:
            events.append({"type": "final", "citation": source})
            return {"status": "completed", "answer": result, "events": events}
    if scenario == "tool-failure":
        events.append({"type": "stop", "reason": "repeated_tool_failure"})
        return {"status": "needs_human", "answer": None, "events": events}
    events.append({"type": "stop", "reason": "no_evidence"})
    return {"status": "unknown", "answer": None, "events": events}


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("scenario", choices=["found-first", "found-second", "missing", "tool-failure", "forbidden-proposal"])
    parser.add_argument("--max-actions", type=int, default=2)
    arguments = parser.parse_args()
    print(json.dumps(run(arguments.scenario, arguments.max_actions), ensure_ascii=False, indent=2))

ls
python react_lab.py found-second

scripted_proposal: search / primary
program_check: allowed=true
action: search / primary
observation: text=null
scripted_proposal: search / secondary
program_check: allowed=true
action: search / secondary
observation: 退款窗口为 7 天；超过窗口需人工审核。
final: citation=secondary
status: completed

python react_lab.py found-first
python react_lab.py missing
python react_lab.py tool-failure
python react_lab.py forbidden-proposal
python react_lab.py found-second --max-actions 1

python -m unittest test_react_lab -v