|
|
@@ -1,27 +1,16 @@
|
|
|
"""demand 示例的最小可运行入口。"""
|
|
|
import asyncio
|
|
|
import copy
|
|
|
+import hashlib
|
|
|
import importlib
|
|
|
import json
|
|
|
import os
|
|
|
import sys
|
|
|
from datetime import datetime
|
|
|
from pathlib import Path
|
|
|
-from typing import Optional
|
|
|
+from typing import Any, Optional
|
|
|
from zoneinfo import ZoneInfo
|
|
|
|
|
|
-from dotenv import load_dotenv
|
|
|
-from sqlalchemy import desc, or_
|
|
|
-
|
|
|
-from examples.demand.changwen_prepare import changwen_prepare
|
|
|
-from examples.demand.config import LOG_LEVEL, ENABLED_TOOLS
|
|
|
-from examples.demand.db_manager import DatabaseManager, query_video_ids_by_names, query_category_level
|
|
|
-from examples.demand.models import TopicPatternExecution
|
|
|
-from examples.demand.piaoquan_prepare import prepare, piaoquan_prepare
|
|
|
-from examples.demand.demand_agent_context import TopicBuildAgentContext
|
|
|
-from examples.demand.mysql import mysql_db
|
|
|
-from examples.demand.zengzhang_prepare import zengzhang_prepare
|
|
|
-
|
|
|
# Clash Verge TUN 模式兼容：禁止 httpx/urllib 自动检测系统 HTTP 代理
|
|
|
os.environ.setdefault("no_proxy", "*")
|
|
|
# 该示例仅使用项目侧能力，禁用框架内置 skills
|
|
|
@@ -32,21 +21,20 @@ os.environ.setdefault("AGENT_STRICT_TOOL_SELECTION", "1")
|
|
|
# 禁用所有侧分支（压缩/反思）
|
|
|
os.environ.setdefault("AGENT_DISABLE_SIDE_BRANCHES", "1")
|
|
|
|
|
|
+from dotenv import load_dotenv
|
|
|
+from sqlalchemy import desc
|
|
|
+
|
|
|
+from examples.demand.db_manager import DatabaseManager, query_video_ids_by_names, query_category_level
|
|
|
+from examples.demand.models import TopicPatternExecution
|
|
|
+from examples.demand.demand_agent_context import TopicBuildAgentContext
|
|
|
+
|
|
|
# 添加项目根目录到 Python 路径
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
|
|
|
|
|
|
load_dotenv()
|
|
|
|
|
|
-from agent.core.runner import AgentRunner
|
|
|
-from agent.llm import create_openrouter_llm_call
|
|
|
-from agent.llm.prompts import SimplePrompt
|
|
|
-from agent.trace import FileSystemTraceStore, Message, Trace
|
|
|
-from agent.utils import setup_logging
|
|
|
from examples.demand.log_capture import build_log, log
|
|
|
|
|
|
-# 导入项目配置
|
|
|
-from examples.demand.config import DEBUG, LOG_FILE, LOG_LEVEL, RUN_CONFIG, TRACE_STORE_PATH
|
|
|
-
|
|
|
CUSTOM_TOOL_MODULES = {
|
|
|
# demand 示例：严格按工具名白名单加载对应模块
|
|
|
"think_and_plan": "examples.demand.agent_tools",
|
|
|
@@ -68,6 +56,80 @@ CUSTOM_TOOL_MODULES = {
|
|
|
"write_execution_summary": "examples.demand.demand_build_agent_tools",
|
|
|
}
|
|
|
|
|
|
+LOCAL_JSON_MODE = "local_json"
|
|
|
+LOCAL_ENTRYPOINT_ENV = "DEMAND_LOCAL_ENTRYPOINT"
|
|
|
+LOCAL_ENTRYPOINT_NAME = "run_existing_execution_local"
|
|
|
+
|
|
|
+
|
|
|
+def _is_local_json_mode() -> bool:
|
|
|
+ return os.getenv("DEMAND_OUTPUT_MODE", "").strip().lower() == LOCAL_JSON_MODE
|
|
|
+
|
|
|
+
|
|
|
+def _require_local_entrypoint() -> None:
|
|
|
+ if not _is_local_json_mode():
|
|
|
+ return
|
|
|
+ entrypoint = os.getenv(LOCAL_ENTRYPOINT_ENV, "").strip()
|
|
|
+ if entrypoint != LOCAL_ENTRYPOINT_NAME:
|
|
|
+ raise RuntimeError(
|
|
|
+ "DEMAND_OUTPUT_MODE=local_json 只能通过 "
|
|
|
+ "examples.demand.run_existing_execution_local 入口运行"
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _default_local_output_root(execution_id: int) -> Path:
|
|
|
+ ts = datetime.now(ZoneInfo("Asia/Shanghai")).strftime("%Y%m%d_%H%M%S")
|
|
|
+ return Path(__file__).parent / "test_output_data" / f"execution_{execution_id}_{ts}"
|
|
|
+
|
|
|
+
|
|
|
+def _ensure_local_output_paths(execution_id: int) -> Path:
|
|
|
+ root_value = os.getenv("DEMAND_LOCAL_OUTPUT_ROOT")
|
|
|
+ root = Path(root_value) if root_value else _default_local_output_root(execution_id)
|
|
|
+ os.environ["DEMAND_LOCAL_OUTPUT_ROOT"] = str(root)
|
|
|
+ os.environ.setdefault("DEMAND_RESULT_BASE_DIR", str(root / "intermediate" / "result"))
|
|
|
+ os.environ.setdefault("DEMAND_TRACE_STORE_PATH", str(root / ".trace"))
|
|
|
+ os.environ.setdefault("DEMAND_OUTPUT_BASE_DIR", str(root / "output"))
|
|
|
+ return root
|
|
|
+
|
|
|
+
|
|
|
+def _get_result_base_dir() -> Path:
|
|
|
+ redirected = os.getenv("DEMAND_RESULT_BASE_DIR")
|
|
|
+ if redirected:
|
|
|
+ return Path(redirected)
|
|
|
+ if _is_local_json_mode():
|
|
|
+ return _ensure_local_output_paths(0) / "result"
|
|
|
+ return Path.cwd() / "result"
|
|
|
+
|
|
|
+
|
|
|
+def _get_output_base_dir() -> Path:
|
|
|
+ redirected = os.getenv("DEMAND_OUTPUT_BASE_DIR")
|
|
|
+ if redirected:
|
|
|
+ return Path(redirected)
|
|
|
+ if _is_local_json_mode():
|
|
|
+ return _ensure_local_output_paths(0) / "output"
|
|
|
+ return Path(__file__).parent / "output"
|
|
|
+
|
|
|
+
|
|
|
+def _get_trace_store_base_dir() -> Path:
|
|
|
+ redirected = os.getenv("DEMAND_TRACE_STORE_PATH")
|
|
|
+ if redirected:
|
|
|
+ return Path(redirected)
|
|
|
+ if _is_local_json_mode():
|
|
|
+ return _ensure_local_output_paths(0) / ".trace"
|
|
|
+ from examples.demand.config import TRACE_STORE_PATH
|
|
|
+
|
|
|
+ return Path(TRACE_STORE_PATH)
|
|
|
+
|
|
|
+
|
|
|
+def _get_local_output_root() -> Optional[Path]:
|
|
|
+ root = os.getenv("DEMAND_LOCAL_OUTPUT_ROOT")
|
|
|
+ return Path(root) if root else None
|
|
|
+
|
|
|
+
|
|
|
+def _write_json_file(path: Path, payload: Any) -> None:
|
|
|
+ path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
+ with open(path, "w", encoding="utf-8") as f:
|
|
|
+ json.dump(payload, f, ensure_ascii=False, indent=2)
|
|
|
+
|
|
|
|
|
|
def get_execution_id_by_merge_level2(cluster_name: str):
|
|
|
"""根据二级品类和平台查询最新的 execution_id。"""
|
|
|
@@ -86,14 +148,14 @@ def get_execution_id_by_merge_level2(cluster_name: str):
|
|
|
session.close()
|
|
|
|
|
|
|
|
|
-def resolve_model(prompt: SimplePrompt) -> str:
|
|
|
+def resolve_model(prompt: Any, run_config: Any) -> str:
|
|
|
model_from_prompt = prompt.config.get("model")
|
|
|
if model_from_prompt:
|
|
|
return model_from_prompt
|
|
|
- return f"anthropic/{RUN_CONFIG.model}" if "/" not in RUN_CONFIG.model else RUN_CONFIG.model
|
|
|
+ return f"anthropic/{run_config.model}" if "/" not in run_config.model else run_config.model
|
|
|
|
|
|
|
|
|
-def extract_assistant_text(message: Message) -> str:
|
|
|
+def extract_assistant_text(message: Any) -> str:
|
|
|
if message.role != "assistant":
|
|
|
return ""
|
|
|
content = message.content
|
|
|
@@ -189,13 +251,238 @@ def _resolve_video_ids_by_name_and_execution_id(name: str, execution_id: int) ->
|
|
|
return query_video_ids_by_names(execution_id=execution_id, names=name_parts)
|
|
|
|
|
|
|
|
|
+def _demand_items_output_path(execution_id: int) -> Path:
|
|
|
+ return _get_result_base_dir() / str(execution_id) / f"execution_id_{execution_id}_demand_items.json"
|
|
|
+
|
|
|
+
|
|
|
+def _resolve_demand_items_path(execution_id: int) -> Optional[Path]:
|
|
|
+ configured_path = _demand_items_output_path(execution_id)
|
|
|
+ if configured_path.exists() or _is_local_json_mode():
|
|
|
+ return configured_path if configured_path.exists() else None
|
|
|
+
|
|
|
+ fallbacks = [
|
|
|
+ Path.cwd() / "result" / str(execution_id) / f"execution_id_{execution_id}_demand_items.json",
|
|
|
+ Path(__file__).parent / "result" / str(execution_id) / f"execution_id_{execution_id}_demand_items.json",
|
|
|
+ ]
|
|
|
+ for path in fallbacks:
|
|
|
+ if path.exists():
|
|
|
+ return path
|
|
|
+ return None
|
|
|
+
|
|
|
+
|
|
|
+def _load_demand_items(execution_id: int, log_prefix: str) -> list[dict]:
|
|
|
+ demand_items_path = _resolve_demand_items_path(execution_id)
|
|
|
+ if not demand_items_path:
|
|
|
+ expected_path = _demand_items_output_path(execution_id)
|
|
|
+ log(f"[{log_prefix}] 未找到需求 JSON：{expected_path}，跳过写入")
|
|
|
+ return []
|
|
|
+
|
|
|
+ try:
|
|
|
+ with open(demand_items_path, "r", encoding="utf-8") as f:
|
|
|
+ loaded = json.load(f)
|
|
|
+ except Exception as e:
|
|
|
+ log(f"[{log_prefix}] 读取需求 JSON 失败：{demand_items_path}，error={e}")
|
|
|
+ return []
|
|
|
+
|
|
|
+ items = loaded["items"] if isinstance(loaded, dict) and isinstance(loaded.get("items"), list) else loaded
|
|
|
+ if not isinstance(items, list):
|
|
|
+ log(f"[{log_prefix}] 需求 JSON 非数组，跳过写入：type={type(items)}")
|
|
|
+ return []
|
|
|
+ return [item for item in items if isinstance(item, dict)]
|
|
|
+
|
|
|
+
|
|
|
+def _normalize_evidence_refs(value: object) -> dict:
|
|
|
+ if isinstance(value, dict):
|
|
|
+ return value
|
|
|
+ if isinstance(value, str) and value.strip():
|
|
|
+ try:
|
|
|
+ loaded = json.loads(value)
|
|
|
+ if isinstance(loaded, dict):
|
|
|
+ return loaded
|
|
|
+ except json.JSONDecodeError:
|
|
|
+ pass
|
|
|
+ return {}
|
|
|
+
|
|
|
+
|
|
|
+def _parse_ext_data(ext_data_raw: object) -> dict:
|
|
|
+ if isinstance(ext_data_raw, dict):
|
|
|
+ return ext_data_raw
|
|
|
+ if isinstance(ext_data_raw, str) and ext_data_raw.strip():
|
|
|
+ try:
|
|
|
+ loaded = json.loads(ext_data_raw)
|
|
|
+ return loaded if isinstance(loaded, dict) else {}
|
|
|
+ except json.JSONDecodeError:
|
|
|
+ return {}
|
|
|
+ return {}
|
|
|
+
|
|
|
+
|
|
|
+def _build_demand_content_rows(
|
|
|
+ execution_id: int,
|
|
|
+ merge_level2: str,
|
|
|
+ *,
|
|
|
+ trace_id: Optional[str] = None,
|
|
|
+ task_id: Optional[int] = None,
|
|
|
+ serialize_ext_data: bool = True,
|
|
|
+ require_evidence: bool = False,
|
|
|
+) -> tuple[list[dict], list[dict], list[dict]]:
|
|
|
+ from examples.demand.evidence_pack_builder import build_evidence_pack
|
|
|
+
|
|
|
+ items = _load_demand_items(execution_id=execution_id, log_prefix="result")
|
|
|
+ if not items:
|
|
|
+ return [], [], []
|
|
|
+
|
|
|
+ dt_value = datetime.now(ZoneInfo("Asia/Shanghai")).strftime("%Y%m%d")
|
|
|
+ score_map = _load_name_score_map(execution_id)
|
|
|
+ rows: list[dict] = []
|
|
|
+ normalized_items: list[dict] = []
|
|
|
+ rejected_items: list[dict] = []
|
|
|
+ for index, di in enumerate(items, start=1):
|
|
|
+ normalized_item = dict(di)
|
|
|
+ evidence_refs = _normalize_evidence_refs(di.get("evidence_refs"))
|
|
|
+ normalized_item["evidence_refs"] = evidence_refs
|
|
|
+ normalized_items.append(normalized_item)
|
|
|
+
|
|
|
+ name = _join_element_names_to_name(di.get("element_names"))
|
|
|
+ if not name:
|
|
|
+ rejected_items.append(
|
|
|
+ {
|
|
|
+ "item_index": index,
|
|
|
+ "reject_reason": "missing element_names/name",
|
|
|
+ "demand_item": normalized_item,
|
|
|
+ }
|
|
|
+ )
|
|
|
+ continue
|
|
|
+
|
|
|
+ score = _avg_score_for_joined_name(name, score_map)
|
|
|
+ reason = di.get("reason")
|
|
|
+ desc_value = di.get("desc")
|
|
|
+ item_type = di.get("type")
|
|
|
+ type_str = str(item_type).strip() if item_type is not None else ""
|
|
|
+ if type_str == "分类":
|
|
|
+ category_level = query_category_level(execution_id=execution_id, name=name)
|
|
|
+ if category_level:
|
|
|
+ type_str = type_str + f"L{category_level}"
|
|
|
+
|
|
|
+ video_ids = _resolve_video_ids_by_name_and_execution_id(name=name, execution_id=execution_id)
|
|
|
+ ext_data: dict[str, Any] = {
|
|
|
+ "reason": reason,
|
|
|
+ "desc": desc_value,
|
|
|
+ "type": type_str,
|
|
|
+ "video_ids": video_ids,
|
|
|
+ }
|
|
|
+ if trace_id:
|
|
|
+ ext_data["trace_id"] = trace_id
|
|
|
+ if task_id is not None:
|
|
|
+ ext_data["demand_task_id"] = task_id
|
|
|
+
|
|
|
+ content_id = len(rows) + 1 if not serialize_ext_data else None
|
|
|
+ if require_evidence or evidence_refs:
|
|
|
+ evidence_result = build_evidence_pack(
|
|
|
+ execution_id=int(execution_id),
|
|
|
+ demand_item=normalized_item,
|
|
|
+ trace_id=trace_id or "",
|
|
|
+ demand_task_id=task_id,
|
|
|
+ demand_content_id=content_id,
|
|
|
+ )
|
|
|
+ if not evidence_result.get("success"):
|
|
|
+ if require_evidence:
|
|
|
+ rejected_items.append(
|
|
|
+ {
|
|
|
+ "item_index": index,
|
|
|
+ "reject_reason": evidence_result.get("reject_reason") or "evidence validation failed",
|
|
|
+ "demand_item": normalized_item,
|
|
|
+ }
|
|
|
+ )
|
|
|
+ continue
|
|
|
+ log(
|
|
|
+ "[result] evidence 校验未通过，生产兼容路径保留旧输出："
|
|
|
+ f"item_index={index}, reason={evidence_result.get('reject_reason')}"
|
|
|
+ )
|
|
|
+ else:
|
|
|
+ evidence_pack = evidence_result["evidence_pack"]
|
|
|
+ ext_data["evidence_pack"] = evidence_pack
|
|
|
+ if evidence_pack.get("video_ids"):
|
|
|
+ ext_data["video_ids"] = evidence_pack["video_ids"]
|
|
|
+
|
|
|
+ row = {
|
|
|
+ "merge_leve2": _safe_truncate(merge_level2, 32),
|
|
|
+ "name": _safe_truncate(name, 64),
|
|
|
+ "reason": reason,
|
|
|
+ "suggestion": desc_value,
|
|
|
+ "score": float(score),
|
|
|
+ "ext_data": json.dumps(ext_data, ensure_ascii=False) if serialize_ext_data else ext_data,
|
|
|
+ "dt": dt_value,
|
|
|
+ }
|
|
|
+ if not serialize_ext_data:
|
|
|
+ row["id"] = content_id
|
|
|
+ rows.append(row)
|
|
|
+
|
|
|
+ if not rows:
|
|
|
+ log("[result] 生成行为空，跳过写入")
|
|
|
+ if rejected_items:
|
|
|
+ log(f"[result] evidence 校验拒绝 {len(rejected_items)} 条 DemandItem")
|
|
|
+ return rows, normalized_items, rejected_items
|
|
|
+
|
|
|
+
|
|
|
+def _build_local_dwd_multi_rows(rows: list[dict]) -> list[dict]:
|
|
|
+ local_rows: list[dict] = []
|
|
|
+ china_today = datetime.now(ZoneInfo("Asia/Shanghai")).strftime("%Y%m%d")
|
|
|
+ for row in rows:
|
|
|
+ merge_leve2 = str(row.get("merge_leve2") or "").strip()
|
|
|
+ name = str(row.get("name") or "").strip()
|
|
|
+ if not merge_leve2 or not name:
|
|
|
+ continue
|
|
|
+
|
|
|
+ ext_data = _parse_ext_data(row.get("ext_data"))
|
|
|
+ type_str = str(ext_data.get("type") or "").strip()
|
|
|
+ video_ids = ext_data.get("video_ids") or []
|
|
|
+ if not isinstance(video_ids, list):
|
|
|
+ video_ids = []
|
|
|
+ video_ids = [str(v).strip() for v in video_ids if v is not None and str(v).strip()]
|
|
|
+ weight = round(float(row.get("score") or 0.0), 6)
|
|
|
+ extend = {"品类": merge_leve2}
|
|
|
+
|
|
|
+ for strategy, demand_name, id_source in [
|
|
|
+ ("当下供需gap", f"{merge_leve2} {name}", f"当下供需gap{merge_leve2} {name}{type_str}{china_today}"),
|
|
|
+ ("当下供需gap-分词", name, f"当下供需gap-分词{name}{merge_leve2}{type_str}{china_today}"),
|
|
|
+ ]:
|
|
|
+ local_rows.append(
|
|
|
+ {
|
|
|
+ "strategy": strategy,
|
|
|
+ "demand_id": hashlib.md5(id_source.encode("utf-8")).hexdigest(),
|
|
|
+ "demand_name": demand_name,
|
|
|
+ "weight": weight,
|
|
|
+ "type": type_str,
|
|
|
+ "video_count": len(video_ids),
|
|
|
+ "video_list": video_ids,
|
|
|
+ "extend": extend,
|
|
|
+ "dt": china_today,
|
|
|
+ }
|
|
|
+ )
|
|
|
+ return local_rows
|
|
|
+
|
|
|
+
|
|
|
+def _build_local_feature_point_rows(names: list[str]) -> list[dict]:
|
|
|
+ dt = datetime.now(ZoneInfo("Asia/Shanghai")).strftime("%Y%m%d")
|
|
|
+ return [
|
|
|
+ {"特征点": name, "总分发曝光pv": 5000, "质bn_rovn": 0.1, "dt": dt}
|
|
|
+ for name in names
|
|
|
+ if name
|
|
|
+ ]
|
|
|
+
|
|
|
+
|
|
|
def _create_demand_task(
|
|
|
execution_id: int,
|
|
|
name: Optional[str] = None,
|
|
|
platform: Optional[str] = None,
|
|
|
) -> Optional[int]:
|
|
|
"""创建 demand_task 记录，返回任务ID。"""
|
|
|
+ if _is_local_json_mode():
|
|
|
+ raise RuntimeError("local_json 模式禁止调用生产 _create_demand_task")
|
|
|
+
|
|
|
try:
|
|
|
+ from examples.demand.mysql import mysql_db
|
|
|
+
|
|
|
# 数据库字段 demand_task.name: varchar(32)
|
|
|
if name is not None:
|
|
|
name = str(name)[:32]
|
|
|
@@ -224,7 +511,11 @@ def _finish_demand_task(task_id: Optional[int], status: int, task_log: str) -> N
|
|
|
"""更新 demand_task 状态与日志。"""
|
|
|
if not task_id:
|
|
|
return
|
|
|
+ if _is_local_json_mode():
|
|
|
+ raise RuntimeError("local_json 模式禁止调用生产 _finish_demand_task")
|
|
|
try:
|
|
|
+ from examples.demand.mysql import mysql_db
|
|
|
+
|
|
|
mysql_db.update(
|
|
|
"demand_task",
|
|
|
{
|
|
|
@@ -239,82 +530,101 @@ def _finish_demand_task(task_id: Optional[int], status: int, task_log: str) -> N
|
|
|
log(f"[task] 更新 demand_task 失败，task_id={task_id}, status={status}, error={e}")
|
|
|
|
|
|
|
|
|
-def write_demand_items_to_mysql(execution_id: int, merge_level2: str) -> int:
|
|
|
- """
|
|
|
- 把 result/{execution_id}/execution_id_{execution_id}_demand_items.json
|
|
|
- 写入 MySQL 表 demand_content
|
|
|
- """
|
|
|
- # create_demand_item(s) 使用 Path.cwd()/result 作为输出目录。
|
|
|
- # 为了兼容“从不同目录启动脚本”的情况，这里同时尝试 cwd 和脚本目录两种结果位置。
|
|
|
- demand_items_path = (
|
|
|
- Path.cwd()
|
|
|
- / "result"
|
|
|
- / str(execution_id)
|
|
|
- / f"execution_id_{execution_id}_demand_items.json"
|
|
|
+def _start_local_demand_task(
|
|
|
+ *,
|
|
|
+ execution_id: int,
|
|
|
+ task_id: Optional[int],
|
|
|
+ name: Optional[str],
|
|
|
+ platform: Optional[str],
|
|
|
+) -> int:
|
|
|
+ from examples.demand.local_output_sink import LocalOutputSink
|
|
|
+
|
|
|
+ local_task_id = int(task_id or os.getenv("DEMAND_LOCAL_TASK_ID", "1"))
|
|
|
+ os.environ["DEMAND_LOCAL_TASK_ID"] = str(local_task_id)
|
|
|
+ root = _get_local_output_root() or _ensure_local_output_paths(execution_id)
|
|
|
+ task_payload = {
|
|
|
+ "id": local_task_id,
|
|
|
+ "execution_id": execution_id,
|
|
|
+ "name": name,
|
|
|
+ "platform": platform,
|
|
|
+ "status": 0,
|
|
|
+ "mode": LOCAL_JSON_MODE,
|
|
|
+ "is_simulated_id": True,
|
|
|
+ "log": "",
|
|
|
+ "started_at": datetime.now(ZoneInfo("Asia/Shanghai")).isoformat(),
|
|
|
+ }
|
|
|
+ LocalOutputSink(root).write_all(
|
|
|
+ run_manifest={
|
|
|
+ "output_mode": LOCAL_JSON_MODE,
|
|
|
+ "execution_id": execution_id,
|
|
|
+ "merge_leve2": name,
|
|
|
+ "platform_type": platform,
|
|
|
+ "task_id": local_task_id,
|
|
|
+ "status": 0,
|
|
|
+ "initialized": True,
|
|
|
+ "created_at": datetime.now(ZoneInfo("Asia/Shanghai")).isoformat(),
|
|
|
+ },
|
|
|
+ demand_task=task_payload,
|
|
|
)
|
|
|
- if not demand_items_path.exists():
|
|
|
- alt_path = (
|
|
|
- Path(__file__).parent
|
|
|
- / "result"
|
|
|
- / str(execution_id)
|
|
|
- / f"execution_id_{execution_id}_demand_items.json"
|
|
|
- )
|
|
|
- if alt_path.exists():
|
|
|
- demand_items_path = alt_path
|
|
|
- else:
|
|
|
- log(f"[mysql] 未找到需求 JSON：{demand_items_path}（也未找到 {alt_path}），跳过写入")
|
|
|
- return 0
|
|
|
+ log(f"[task][local_json] 写入本地 demand_task，task_id={local_task_id}, execution_id={execution_id}")
|
|
|
+ return local_task_id
|
|
|
|
|
|
- try:
|
|
|
- with open(demand_items_path, "r", encoding="utf-8") as f:
|
|
|
- loaded = json.load(f)
|
|
|
- except Exception as e:
|
|
|
- log(f"[mysql] 读取需求 JSON 失败：{demand_items_path}，error={e}")
|
|
|
- return 0
|
|
|
|
|
|
- items = loaded["items"] if isinstance(loaded, dict) and isinstance(loaded.get("items"), list) else loaded
|
|
|
- if not isinstance(items, list):
|
|
|
- log(f"[mysql] 需求 JSON 非数组，跳过写入：type={type(items)}")
|
|
|
- return 0
|
|
|
+def _finish_local_demand_task(task_id: Optional[int], status: int, task_log: str) -> None:
|
|
|
+ if not task_id:
|
|
|
+ return
|
|
|
+ root = _get_local_output_root()
|
|
|
+ if not root:
|
|
|
+ return
|
|
|
+ task_path = root / "demand_task.json"
|
|
|
+ payload: dict[str, Any] = {}
|
|
|
+ if task_path.exists():
|
|
|
+ try:
|
|
|
+ with open(task_path, "r", encoding="utf-8") as f:
|
|
|
+ loaded = json.load(f)
|
|
|
+ if isinstance(loaded, dict):
|
|
|
+ payload = loaded
|
|
|
+ except Exception:
|
|
|
+ payload = {}
|
|
|
+ payload.update(
|
|
|
+ {
|
|
|
+ "id": task_id,
|
|
|
+ "status": int(status),
|
|
|
+ "mode": LOCAL_JSON_MODE,
|
|
|
+ "is_simulated_id": True,
|
|
|
+ "log": task_log or "",
|
|
|
+ "finished_at": datetime.now(ZoneInfo("Asia/Shanghai")).isoformat(),
|
|
|
+ }
|
|
|
+ )
|
|
|
+ _write_json_file(task_path, payload)
|
|
|
+ log(f"[task][local_json] 更新本地 demand_task，task_id={task_id}, status={status}")
|
|
|
|
|
|
- dt_value = datetime.now(ZoneInfo("Asia/Shanghai")).strftime("%Y%m%d")
|
|
|
- score_map = _load_name_score_map(execution_id)
|
|
|
- rows: list[dict] = []
|
|
|
- for di in items:
|
|
|
- if not isinstance(di, dict):
|
|
|
- continue
|
|
|
|
|
|
- name = _join_element_names_to_name(di.get("element_names"))
|
|
|
- if not name:
|
|
|
- continue
|
|
|
- score = _avg_score_for_joined_name(name, score_map)
|
|
|
- reason = di.get("reason")
|
|
|
- desc_value = di.get("desc")
|
|
|
- type = di.get("type")
|
|
|
- suggestion = desc_value
|
|
|
- type_str = str(type).strip() if type is not None else ""
|
|
|
- if type_str == "分类":
|
|
|
- category_level = query_category_level(execution_id=execution_id, name=name)
|
|
|
- if category_level:
|
|
|
- type_str = type_str + f"L{category_level}"
|
|
|
- video_ids = _resolve_video_ids_by_name_and_execution_id(name=name, execution_id=execution_id)
|
|
|
- # 兼容旧字段：同时保留 ext_data（reason/desc）JSON，便于旧版消费逻辑迁移期继续使用。
|
|
|
- ext_data = {"reason": reason, "desc": desc_value, "type": type_str, "video_ids": video_ids}
|
|
|
+def write_demand_items_to_mysql(
|
|
|
+ execution_id: int,
|
|
|
+ merge_level2: str,
|
|
|
+ *,
|
|
|
+ trace_id: Optional[str] = None,
|
|
|
+ task_id: Optional[int] = None,
|
|
|
+) -> int:
|
|
|
+ """
|
|
|
+ 把 result/{execution_id}/execution_id_{execution_id}_demand_items.json
|
|
|
+ 写入 MySQL 表 demand_content
|
|
|
+ """
|
|
|
+ if _is_local_json_mode():
|
|
|
+ raise RuntimeError("local_json 模式禁止调用生产 write_demand_items_to_mysql")
|
|
|
|
|
|
- rows.append(
|
|
|
- {
|
|
|
- "merge_leve2": _safe_truncate(merge_level2, 32),
|
|
|
- "name": _safe_truncate(name, 64),
|
|
|
- "reason": reason,
|
|
|
- "suggestion": suggestion,
|
|
|
- "score": float(score),
|
|
|
- "ext_data": json.dumps(ext_data, ensure_ascii=False),
|
|
|
- "dt": dt_value,
|
|
|
- }
|
|
|
- )
|
|
|
+ from examples.demand.mysql import mysql_db
|
|
|
|
|
|
+ rows, _, rejected_items = _build_demand_content_rows(
|
|
|
+ execution_id=execution_id,
|
|
|
+ merge_level2=merge_level2,
|
|
|
+ trace_id=trace_id,
|
|
|
+ task_id=task_id,
|
|
|
+ )
|
|
|
+ if rejected_items:
|
|
|
+ log(f"[mysql] evidence 校验拒绝 rows={len(rejected_items)}，仅写入 passed rows")
|
|
|
if not rows:
|
|
|
- log("[mysql] 生成行为空，跳过写入")
|
|
|
return 0
|
|
|
|
|
|
affected = mysql_db.insert_many("demand_content", rows)
|
|
|
@@ -324,6 +634,7 @@ def write_demand_items_to_mysql(execution_id: int, merge_level2: str) -> int:
|
|
|
from examples.demand.data_query_tools import write_dwd_multi_demand_pool_di_to_hive
|
|
|
|
|
|
hive_written = write_dwd_multi_demand_pool_di_to_hive(rows=rows)
|
|
|
+ dt_value = datetime.now(ZoneInfo("Asia/Shanghai")).strftime("%Y%m%d")
|
|
|
log(f"[hive] 写入 dwd_multi_demand_pool_di 完成，rows={hive_written}, dt={dt_value}")
|
|
|
except Exception as e:
|
|
|
log(f"[hive] 写入 dwd_multi_demand_pool_di 异常（MySQL 已成功）：{e}")
|
|
|
@@ -333,40 +644,49 @@ def write_demand_items_to_mysql(execution_id: int, merge_level2: str) -> int:
|
|
|
return len(rows)
|
|
|
|
|
|
|
|
|
+def write_demand_items_to_local_json(
|
|
|
+ execution_id: int,
|
|
|
+ merge_level2: str,
|
|
|
+ *,
|
|
|
+ trace_id: Optional[str] = None,
|
|
|
+ task_id: Optional[int] = None,
|
|
|
+) -> int:
|
|
|
+ """把普通需求池输出镜像到本地 JSON，不写 demand_content / Hive。"""
|
|
|
+ from examples.demand.local_output_sink import LocalOutputSink
|
|
|
+
|
|
|
+ root = _get_local_output_root() or _ensure_local_output_paths(execution_id)
|
|
|
+ rows, demand_items, rejected_items = _build_demand_content_rows(
|
|
|
+ execution_id=execution_id,
|
|
|
+ merge_level2=merge_level2,
|
|
|
+ trace_id=trace_id,
|
|
|
+ task_id=task_id,
|
|
|
+ serialize_ext_data=False,
|
|
|
+ require_evidence=True,
|
|
|
+ )
|
|
|
+
|
|
|
+ sink = LocalOutputSink(root)
|
|
|
+ sink.write_demand_items(demand_items)
|
|
|
+ sink.write_rejected_demand_items(rejected_items)
|
|
|
+ sink.write_demand_content(rows)
|
|
|
+ sink.write_dwd_multi_demand_pool_di_from_demand_content(rows)
|
|
|
+ sink.write_feature_point_data([])
|
|
|
+ log(
|
|
|
+ "[local_json] 写入本地 demand_content/dwd_multi_demand_pool_di 完成，"
|
|
|
+ f"demand_content_rows={len(rows)}, rejected={len(rejected_items)}, root={root}"
|
|
|
+ )
|
|
|
+ return len(rows)
|
|
|
+
|
|
|
+
|
|
|
def write_global_tree_demand_items_to_hive(execution_id: int, merge_level2: str) -> int:
|
|
|
"""
|
|
|
把 result/{execution_id}/execution_id_{execution_id}_demand_items.json
|
|
|
写入 Hive 表 feature_point_data（仅全局树场景）。
|
|
|
"""
|
|
|
- demand_items_path = (
|
|
|
- Path.cwd()
|
|
|
- / "result"
|
|
|
- / str(execution_id)
|
|
|
- / f"execution_id_{execution_id}_demand_items.json"
|
|
|
- )
|
|
|
- if not demand_items_path.exists():
|
|
|
- alt_path = (
|
|
|
- Path(__file__).parent
|
|
|
- / "result"
|
|
|
- / str(execution_id)
|
|
|
- / f"execution_id_{execution_id}_demand_items.json"
|
|
|
- )
|
|
|
- if alt_path.exists():
|
|
|
- demand_items_path = alt_path
|
|
|
- else:
|
|
|
- log(f"[hive] 未找到需求 JSON：{demand_items_path}（也未找到 {alt_path}），跳过写入")
|
|
|
- return 0
|
|
|
+ if _is_local_json_mode():
|
|
|
+ raise RuntimeError("local_json 模式禁止调用生产 write_global_tree_demand_items_to_hive")
|
|
|
|
|
|
- try:
|
|
|
- with open(demand_items_path, "r", encoding="utf-8") as f:
|
|
|
- loaded = json.load(f)
|
|
|
- except Exception as e:
|
|
|
- log(f"[hive] 读取需求 JSON 失败：{demand_items_path}，error={e}")
|
|
|
- return 0
|
|
|
-
|
|
|
- items = loaded["items"] if isinstance(loaded, dict) and isinstance(loaded.get("items"), list) else loaded
|
|
|
- if not isinstance(items, list):
|
|
|
- log(f"[hive] 需求 JSON 非数组，跳过写入：type={type(items)}")
|
|
|
+ items = _load_demand_items(execution_id=execution_id, log_prefix="hive")
|
|
|
+ if not items:
|
|
|
return 0
|
|
|
|
|
|
names: list[str] = []
|
|
|
@@ -393,24 +713,123 @@ def write_global_tree_demand_items_to_hive(execution_id: int, merge_level2: str)
|
|
|
return 0
|
|
|
|
|
|
|
|
|
-async def run_once(execution_id, merge_level2, count: int = 30, task_id: Optional[int] = None) -> str:
|
|
|
+def write_global_tree_demand_items_to_local_json(
|
|
|
+ execution_id: int,
|
|
|
+ merge_level2: str,
|
|
|
+ *,
|
|
|
+ trace_id: Optional[str] = None,
|
|
|
+ task_id: Optional[int] = None,
|
|
|
+) -> int:
|
|
|
+ """把全局树 feature_point_data 输出镜像到本地 JSON，不写 Hive。"""
|
|
|
+ from examples.demand.local_output_sink import LocalOutputSink
|
|
|
+
|
|
|
+ root = _get_local_output_root() or _ensure_local_output_paths(execution_id)
|
|
|
+ items = _load_demand_items(execution_id=execution_id, log_prefix="local_json")
|
|
|
+ normalized_items = []
|
|
|
+ names: list[str] = []
|
|
|
+ for di in items:
|
|
|
+ normalized = dict(di)
|
|
|
+ normalized["evidence_refs"] = _normalize_evidence_refs(di.get("evidence_refs"))
|
|
|
+ normalized_items.append(normalized)
|
|
|
+ name = _join_element_names_to_name(di.get("element_names"))
|
|
|
+ if name:
|
|
|
+ names.append(name)
|
|
|
+
|
|
|
+ sink = LocalOutputSink(root)
|
|
|
+ sink.write_demand_items(normalized_items)
|
|
|
+ sink.write_rejected_demand_items([])
|
|
|
+ sink.write_demand_content([])
|
|
|
+ sink.write_dwd_multi_demand_pool_di([])
|
|
|
+ sink.write_feature_point_data_from_names(names)
|
|
|
+ log(
|
|
|
+ "[local_json] 写入本地 feature_point_data 完成，"
|
|
|
+ f"rows={len(names)}, merge_leve2={merge_level2}, root={root}"
|
|
|
+ )
|
|
|
+ return len(names)
|
|
|
+
|
|
|
+
|
|
|
+def _write_local_run_manifest(
|
|
|
+ *,
|
|
|
+ execution_id: int,
|
|
|
+ merge_level2: str,
|
|
|
+ platform_type: Optional[str],
|
|
|
+ count: int,
|
|
|
+ task_id: Optional[int],
|
|
|
+ trace_id: Optional[str],
|
|
|
+ final_text: str,
|
|
|
+ task_status: int,
|
|
|
+ total_tokens: int,
|
|
|
+ total_cost: float,
|
|
|
+ result_write_count: int,
|
|
|
+ log_file_path: Path,
|
|
|
+ result_write_error: Optional[str] = None,
|
|
|
+) -> None:
|
|
|
+ if not _is_local_json_mode():
|
|
|
+ return
|
|
|
+ root = _get_local_output_root() or _ensure_local_output_paths(execution_id)
|
|
|
+ manifest = {
|
|
|
+ "output_mode": LOCAL_JSON_MODE,
|
|
|
+ "execution_id": execution_id,
|
|
|
+ "merge_leve2": merge_level2,
|
|
|
+ "platform_type": platform_type,
|
|
|
+ "count": count,
|
|
|
+ "task_id": task_id,
|
|
|
+ "trace_id": trace_id,
|
|
|
+ "status": task_status,
|
|
|
+ "total_tokens": total_tokens,
|
|
|
+ "total_cost": total_cost,
|
|
|
+ "result_write_count": result_write_count,
|
|
|
+ "result_write_error": result_write_error,
|
|
|
+ "final_text_length": len(final_text or ""),
|
|
|
+ "paths": {
|
|
|
+ "root": str(root),
|
|
|
+ "result": str(_get_result_base_dir()),
|
|
|
+ "trace": str(_get_trace_store_base_dir()),
|
|
|
+ "output": str(_get_output_base_dir()),
|
|
|
+ "demand_items_raw": str(_demand_items_output_path(execution_id)),
|
|
|
+ "log_file": str(log_file_path),
|
|
|
+ },
|
|
|
+ "created_at": datetime.now(ZoneInfo("Asia/Shanghai")).isoformat(),
|
|
|
+ }
|
|
|
+ _write_json_file(root / "run_manifest.json", manifest)
|
|
|
+
|
|
|
+
|
|
|
+async def run_once(
|
|
|
+ execution_id,
|
|
|
+ merge_level2,
|
|
|
+ count: int = 30,
|
|
|
+ task_id: Optional[int] = None,
|
|
|
+ platform_type: Optional[str] = None,
|
|
|
+) -> dict:
|
|
|
+ from agent.core.runner import AgentRunner
|
|
|
+ from agent.llm import create_openrouter_llm_call
|
|
|
+ from agent.llm.prompts import SimplePrompt
|
|
|
+ from agent.trace import FileSystemTraceStore, Message, Trace
|
|
|
+ from agent.utils import setup_logging
|
|
|
+ from examples.demand.config import DEBUG, ENABLED_TOOLS, LOG_FILE, LOG_LEVEL, RUN_CONFIG
|
|
|
+
|
|
|
task_log_text = ""
|
|
|
task_status = 0
|
|
|
+ result_write_count = 0
|
|
|
+ result_write_error: Optional[str] = None
|
|
|
+
|
|
|
+ if _is_local_json_mode():
|
|
|
+ _ensure_local_output_paths(execution_id)
|
|
|
|
|
|
TopicBuildAgentContext.set_execution_id(execution_id)
|
|
|
+ TopicBuildAgentContext.set_metadata("result_base_dir", str(_get_result_base_dir()))
|
|
|
|
|
|
base_dir = Path(__file__).parent
|
|
|
- output_dir = base_dir / "output"
|
|
|
- output_dir.mkdir(exist_ok=True)
|
|
|
+ output_dir = _get_output_base_dir()
|
|
|
+ output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
setup_logging(level=LOG_LEVEL, file=LOG_FILE)
|
|
|
register_selected_tools(ENABLED_TOOLS)
|
|
|
|
|
|
prompt = SimplePrompt(base_dir / "demand.md")
|
|
|
|
|
|
- model = resolve_model(prompt)
|
|
|
-
|
|
|
run_config = copy.deepcopy(RUN_CONFIG)
|
|
|
+ model = resolve_model(prompt, run_config)
|
|
|
run_config.temperature = float(prompt.config.get("temperature", run_config.temperature))
|
|
|
run_config.max_iterations = int(prompt.config.get("max_iterations", run_config.max_iterations))
|
|
|
run_config.tools = ENABLED_TOOLS.copy()
|
|
|
@@ -425,7 +844,8 @@ async def run_once(execution_id, merge_level2, count: int = 30, task_id: Optiona
|
|
|
|
|
|
initial_messages = prompt.build_messages(merge_level2=merge_level2, count=count)
|
|
|
|
|
|
- store = FileSystemTraceStore(base_path=TRACE_STORE_PATH)
|
|
|
+ trace_store_path = _get_trace_store_base_dir()
|
|
|
+ store = FileSystemTraceStore(base_path=str(trace_store_path))
|
|
|
runner = AgentRunner(
|
|
|
trace_store=store,
|
|
|
llm_call=create_openrouter_llm_call(model=model),
|
|
|
@@ -436,6 +856,7 @@ async def run_once(execution_id, merge_level2, count: int = 30, task_id: Optiona
|
|
|
final_text = ""
|
|
|
total_tokens = 0
|
|
|
total_cost = 0.0
|
|
|
+ trace_id: Optional[str] = None
|
|
|
has_completed_trace_cost = False
|
|
|
log_file_path = output_dir / f"{execution_id}" / f"run_log_{execution_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
|
|
|
log_file_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
@@ -443,6 +864,7 @@ async def run_once(execution_id, merge_level2, count: int = 30, task_id: Optiona
|
|
|
with build_log(execution_id) as log_buffer:
|
|
|
async for item in runner.run(messages=initial_messages, config=run_config):
|
|
|
if isinstance(item, Trace):
|
|
|
+ trace_id = getattr(item, "trace_id", None) or trace_id
|
|
|
if getattr(item, "status", None) == "completed":
|
|
|
total_tokens = int(getattr(item, "total_tokens", 0) or 0)
|
|
|
total_cost = float(getattr(item, "total_cost", 0.0) or 0.0)
|
|
|
@@ -468,15 +890,43 @@ async def run_once(execution_id, merge_level2, count: int = 30, task_id: Optiona
|
|
|
# agent 执行完成后：全局树写 Hive，其他写 MySQL
|
|
|
try:
|
|
|
if str(merge_level2).strip() == "全局树":
|
|
|
- write_global_tree_demand_items_to_hive(execution_id=execution_id, merge_level2=merge_level2)
|
|
|
+ if _is_local_json_mode():
|
|
|
+ result_write_count = write_global_tree_demand_items_to_local_json(
|
|
|
+ execution_id=execution_id,
|
|
|
+ merge_level2=merge_level2,
|
|
|
+ trace_id=trace_id,
|
|
|
+ task_id=task_id,
|
|
|
+ )
|
|
|
+ else:
|
|
|
+ result_write_count = write_global_tree_demand_items_to_hive(
|
|
|
+ execution_id=execution_id,
|
|
|
+ merge_level2=merge_level2,
|
|
|
+ )
|
|
|
else:
|
|
|
# element_names -> name（逗号分隔）；reason -> demand_content.reason；desc -> demand_content.suggestion；dt -> demand_content.dt
|
|
|
- write_demand_items_to_mysql(execution_id=execution_id, merge_level2=merge_level2)
|
|
|
+ if _is_local_json_mode():
|
|
|
+ result_write_count = write_demand_items_to_local_json(
|
|
|
+ execution_id=execution_id,
|
|
|
+ merge_level2=merge_level2,
|
|
|
+ trace_id=trace_id,
|
|
|
+ task_id=task_id,
|
|
|
+ )
|
|
|
+ else:
|
|
|
+ result_write_count = write_demand_items_to_mysql(
|
|
|
+ execution_id=execution_id,
|
|
|
+ merge_level2=merge_level2,
|
|
|
+ trace_id=trace_id,
|
|
|
+ task_id=task_id,
|
|
|
+ )
|
|
|
except Exception as e:
|
|
|
+ result_write_error = str(e)
|
|
|
log(f"[result-write] 写入结果异常：{e}")
|
|
|
+ if _is_local_json_mode():
|
|
|
+ task_status = 2
|
|
|
|
|
|
task_log_text = log_buffer.getvalue()
|
|
|
- task_status = 1
|
|
|
+ if task_status != 2:
|
|
|
+ task_status = 1
|
|
|
except Exception as e:
|
|
|
if not task_log_text:
|
|
|
# 如果异常发生在 build_log 内部，尽量回收已产生的日志
|
|
|
@@ -498,9 +948,34 @@ async def run_once(execution_id, merge_level2, count: int = 30, task_id: Optiona
|
|
|
except Exception:
|
|
|
# 兜底：即使写文件失败，也要确保 MySQL 状态被更新
|
|
|
pass
|
|
|
- _finish_demand_task(task_id=task_id, status=task_status, task_log=task_log_text)
|
|
|
+ if _is_local_json_mode():
|
|
|
+ _finish_local_demand_task(task_id=task_id, status=task_status, task_log=task_log_text)
|
|
|
+ else:
|
|
|
+ _finish_demand_task(task_id=task_id, status=task_status, task_log=task_log_text)
|
|
|
+ _write_local_run_manifest(
|
|
|
+ execution_id=execution_id,
|
|
|
+ merge_level2=merge_level2,
|
|
|
+ platform_type=platform_type,
|
|
|
+ count=int(count),
|
|
|
+ task_id=task_id,
|
|
|
+ trace_id=trace_id,
|
|
|
+ final_text=final_text,
|
|
|
+ task_status=task_status,
|
|
|
+ total_tokens=total_tokens,
|
|
|
+ total_cost=total_cost,
|
|
|
+ result_write_count=result_write_count,
|
|
|
+ log_file_path=log_file_path,
|
|
|
+ result_write_error=result_write_error,
|
|
|
+ )
|
|
|
|
|
|
- return final_text
|
|
|
+ return {
|
|
|
+ "final_text": final_text,
|
|
|
+ "trace_id": trace_id,
|
|
|
+ "output_dir": str(output_dir / f"{execution_id}"),
|
|
|
+ "trace_store_path": str(trace_store_path),
|
|
|
+ "result_base_dir": str(_get_result_base_dir()),
|
|
|
+ "result_write_count": result_write_count,
|
|
|
+ }
|
|
|
|
|
|
|
|
|
async def main(
|
|
|
@@ -510,20 +985,58 @@ async def main(
|
|
|
execution_id: Optional[int] = None,
|
|
|
task_id: Optional[int] = None,
|
|
|
) -> dict:
|
|
|
+ if _is_local_json_mode():
|
|
|
+ _require_local_entrypoint()
|
|
|
+ if execution_id is None:
|
|
|
+ raise ValueError("local_json 模式必须传入已有 execution_id，不能触发 prepare 写库流程")
|
|
|
+ _ensure_local_output_paths(execution_id)
|
|
|
+
|
|
|
if execution_id is None:
|
|
|
if platform_type == "piaoquan":
|
|
|
+ from examples.demand.piaoquan_prepare import piaoquan_prepare
|
|
|
+
|
|
|
execution_id = piaoquan_prepare(cluster_name)
|
|
|
elif platform_type == "changwen":
|
|
|
+ from examples.demand.changwen_prepare import changwen_prepare
|
|
|
+
|
|
|
execution_id = changwen_prepare(cluster_name)
|
|
|
elif platform_type == "zengzhang":
|
|
|
+ from examples.demand.zengzhang_prepare import zengzhang_prepare
|
|
|
+
|
|
|
execution_id = zengzhang_prepare(cluster_name)
|
|
|
else:
|
|
|
execution_id = None
|
|
|
if not execution_id:
|
|
|
- return {"execution_id": None, "final_text": ""}
|
|
|
+ return {"execution_id": None, "trace_id": None, "final_text": ""}
|
|
|
+
|
|
|
+ if _is_local_json_mode():
|
|
|
+ task_id = _start_local_demand_task(
|
|
|
+ execution_id=int(execution_id),
|
|
|
+ task_id=task_id,
|
|
|
+ name=cluster_name,
|
|
|
+ platform=platform_type,
|
|
|
+ )
|
|
|
|
|
|
- final_text = await run_once(execution_id, cluster_name, count=count, task_id=task_id)
|
|
|
- return {"execution_id": execution_id, "final_text": final_text}
|
|
|
+ run_result = await run_once(
|
|
|
+ execution_id,
|
|
|
+ cluster_name,
|
|
|
+ count=count,
|
|
|
+ task_id=task_id,
|
|
|
+ platform_type=platform_type,
|
|
|
+ )
|
|
|
+ result = {
|
|
|
+ "execution_id": execution_id,
|
|
|
+ "trace_id": run_result.get("trace_id"),
|
|
|
+ "final_text": run_result.get("final_text", ""),
|
|
|
+ "output_dir": run_result.get("output_dir"),
|
|
|
+ "trace_store_path": run_result.get("trace_store_path"),
|
|
|
+ "result_base_dir": run_result.get("result_base_dir"),
|
|
|
+ "result_write_count": run_result.get("result_write_count", 0),
|
|
|
+ }
|
|
|
+ if _is_local_json_mode():
|
|
|
+ root = _get_local_output_root()
|
|
|
+ result["local_output_root"] = str(root) if root else None
|
|
|
+ return result
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
SamLee