demand-agent-new/examples/demand/demand_build_agent_tools.py

import json
import os
from pathlib import Path
from typing import Any, Dict, List, Optional

from agent import tool
from examples.demand.demand_agent_context import TopicBuildAgentContext
from examples.demand.tool_logging import log_tool_input, log_tool_output


def _get_result_base_dir() -> Path:
    """输出到结果目录；local_json 模式可通过环境变量重定向。"""
    metadata_dir = TopicBuildAgentContext.get_metadata("result_base_dir")
    if metadata_dir:
        return Path(metadata_dir)
    redirected = os.getenv("DEMAND_RESULT_BASE_DIR")
    if redirected:
        return Path(redirected)
    return Path.cwd() / "result"


def _normalize_evidence_refs(evidence_refs: Any) -> Dict[str, Any]:
    if evidence_refs is None:
        return {}
    if isinstance(evidence_refs, dict):
        return evidence_refs
    if isinstance(evidence_refs, str) and evidence_refs.strip():
        try:
            loaded = json.loads(evidence_refs)
            if isinstance(loaded, dict):
                return loaded
        except json.JSONDecodeError:
            pass
    return {"raw": evidence_refs}


@tool(
    "存储需求到结果集。 - element_names - reason（原因）- desc（需求描述）- type（来源类型）- evidence_refs（候选证据引用）"
)
def create_demand_item(
        element_names: List[str] = None,
        reason: str = None,
        desc: str = None,
        type: str = None,
        evidence_refs: Optional[Dict[str, Any]] = None) -> str:
    """
    每次调用向“execution_id 对应的本地 JSON 文件”追加一条记录。

    写入对象包含以下字段：
      - element_names
      - reason（原因）
      - desc（需求描述）
      - type（来源类型）
      - evidence_refs（候选证据引用，后续由 orchestration/校验逻辑决定是否能升级为 evidence_pack）
    """
    execution_id: Optional[int] = TopicBuildAgentContext.get_execution_id()
    normalized_evidence_refs = _normalize_evidence_refs(evidence_refs)
    params: Dict[str, Any] = {
        "execution_id": execution_id,
        "element_names": element_names,
        "reason": reason,
        "desc": desc,
        "type": type,
        "evidence_refs": normalized_evidence_refs,
    }
    log_tool_input("create_demand_item", params)

    if not execution_id:
        return log_tool_output("create_demand_item", "错误: 未设置 execution_id")

    record: Dict[str, Any] = {
        "element_names": element_names,
        "reason": reason,
        "desc": desc,
        "type": type,
        "evidence_refs": normalized_evidence_refs,
    }

    # 按 execution_id 区分文件，避免不同执行互相污染。
    # 例如：result/{execution_id}/execution_id_{execution_id}_demand_items.json
    output_dir = _get_result_base_dir() / f"{execution_id}"
    output_path = output_dir / f"execution_id_{execution_id}_demand_items.json"
    output_path.parent.mkdir(parents=True, exist_ok=True)

    items: List[Dict[str, Any]] = []
    if output_path.exists():
        try:
            with open(output_path, "r", encoding="utf-8") as f:
                loaded = json.load(f)
            if isinstance(loaded, list):
                items = loaded
            elif isinstance(loaded, dict) and isinstance(loaded.get("items"), list):
                # 兼容可能的包装格式：{"items":[...]}
                items = loaded["items"]
            else:
                # 兜底：把已有内容当作单条记录追加
                items = [loaded]
        except json.JSONDecodeError:
            # 文件内容损坏时，不阻断执行；从空列表开始追加
            items = []

    items.append(record)
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(items, f, ensure_ascii=False, indent=2)

    result = json.dumps(
        {"success": True, "execution_id": execution_id, "written_to": str(output_path)},
        ensure_ascii=False,
    )
    return log_tool_output("create_demand_item", result)


@tool(
    "批量存储需求到结果集。 - element_names - reason（原因）- desc（需求描述）- type（来源类型）- evidence_refs（候选证据引用）"
)
def create_demand_items(demand_items: List[Dict[str, Any]] = None) -> str:
    """
    一次调用追加多条记录到“execution_id 对应的本地 JSON 文件”（JSON 数组）。

    每条记录字段：
      - element_names
      - reason（原因）
      - desc（需求描述）
      - type（来源类型）
      - evidence_refs（候选证据引用）
    """
    execution_id: Optional[int] = TopicBuildAgentContext.get_execution_id()
    params: Dict[str, Any] = {"execution_id": execution_id, "count": len(demand_items or []),
                              "demand_items": demand_items}
    log_tool_input("create_demand_items", params)

    if not execution_id:
        return log_tool_output("create_demand_items", "错误: 未设置 execution_id")

    if not demand_items or not isinstance(demand_items, list):
        return log_tool_output("create_demand_items", "错误: demand_items 必须为非空列表")

    output_dir = _get_result_base_dir() / f"{execution_id}"
    output_path = output_dir / f"execution_id_{execution_id}_demand_items.json"
    output_path.parent.mkdir(parents=True, exist_ok=True)

    items: List[Dict[str, Any]] = []
    if output_path.exists():
        try:
            with open(output_path, "r", encoding="utf-8") as f:
                loaded = json.load(f)
            if isinstance(loaded, list):
                items = loaded
            elif isinstance(loaded, dict) and isinstance(loaded.get("items"), list):
                items = loaded["items"]
            else:
                items = [loaded]
        except json.JSONDecodeError:
            items = []

    written_records: List[Dict[str, Any]] = []
    for i, di in enumerate(demand_items):
        if not isinstance(di, dict):
            return log_tool_output("create_demand_items", f"错误: demand_items[{i}] 必须为对象(dict)")
        record = {
            "element_names": di.get("element_names"),
            "reason": di.get("reason"),
            "desc": di.get("desc"),
            "type": di.get("type"),
            "evidence_refs": _normalize_evidence_refs(di.get("evidence_refs")),
        }
        written_records.append(record)

    items.extend(written_records)
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(items, f, ensure_ascii=False, indent=2)

    result = json.dumps(
        {
            "success": True,
            "execution_id": execution_id,
            "written_to": str(output_path),
            "written_count": len(written_records),
        },
        ensure_ascii=False,
    )
    return log_tool_output("create_demand_items", result)


@tool(
    "写入本次执行总结（在所有任务执行完成后调用）。"
    "\n\n该工具用于把最终总结记录到本地/trace输出中（框架侧通过返回值与日志落盘）。"
)
def write_execution_summary(summary: str) -> str:
    """写入本次执行总结。在所有任务执行完成后调用。

    Args:
        summary: 执行总结（Markdown 格式）。

    Returns:
        JSON 字符串：
        - 成功：`{"success": True, "execution_id": execution_id}`
        - 失败：`"错误: 未设置 execution_id"`
    """
    execution_id: Optional[int] = TopicBuildAgentContext.get_execution_id()
    params: Dict[str, str] = {"summary": summary}
    log_tool_input("write_execution_summary", params)
    if not execution_id:
        return log_tool_output("write_execution_summary", "错误: 未设置 execution_id")
    result = json.dumps({"success": True, "execution_id": execution_id}, ensure_ascii=False)
    return log_tool_output("write_execution_summary", result)