"""demand 示例的最小可运行入口。""" import asyncio import copy import importlib import json import os import sys from datetime import datetime from pathlib import Path from typing import Any, Optional from zoneinfo import ZoneInfo # Clash Verge TUN 模式兼容：禁止 httpx/urllib 自动检测系统 HTTP 代理 os.environ.setdefault("no_proxy", "*") # 该示例仅使用项目侧能力，禁用框架内置 skills os.environ.setdefault("AGENT_DISABLE_BUILTIN_SKILLS", "1") # 禁用内置工具自动注册，并开启严格工具白名单 os.environ.setdefault("AGENT_DISABLE_BUILTIN_TOOL_REGISTRATION", "1") os.environ.setdefault("AGENT_STRICT_TOOL_SELECTION", "1") # 禁用所有侧分支（压缩/反思） os.environ.setdefault("AGENT_DISABLE_SIDE_BRANCHES", "1") from dotenv import load_dotenv from examples.demand.db_manager import ( DEFAULT_DEMAND_PLATFORM, normalize_platform, query_category_level, query_latest_success_execution_id, query_video_ids_by_names, ) from examples.demand.demand_agent_context import TopicBuildAgentContext # 添加项目根目录到 Python 路径 sys.path.insert(0, str(Path(__file__).parent.parent.parent)) load_dotenv() from examples.demand.log_capture import build_log, log CUSTOM_TOOL_MODULES = { # demand 示例：严格按工具名白名单加载对应模块 "think_and_plan": "examples.demand.agent_tools", "get_category_tree": "examples.demand.demand_pattern_tools", "get_frequent_itemsets": "examples.demand.demand_pattern_tools", "get_itemset_detail": "examples.demand.demand_pattern_tools", "get_post_elements": "examples.demand.demand_pattern_tools", "search_elements": "examples.demand.demand_pattern_tools", "get_element_category_chain": "examples.demand.demand_pattern_tools", "get_category_detail": "examples.demand.demand_pattern_tools", "search_categories": "examples.demand.demand_pattern_tools", "get_category_elements": "examples.demand.demand_pattern_tools", "get_category_co_occurrences": "examples.demand.demand_pattern_tools", "get_element_co_occurrences": "examples.demand.demand_pattern_tools", "get_weight_score_topn": "examples.demand.weight_score_query_tools", "get_weight_score_by_name": "examples.demand.weight_score_query_tools", "create_demand_item": "examples.demand.demand_build_agent_tools", "create_demand_items": "examples.demand.demand_build_agent_tools", "write_execution_summary": "examples.demand.demand_build_agent_tools", } LOCAL_JSON_MODE = "local_json" MYSQL_DEMAND_CONTENT_MODE = "mysql_demand_content" LOCAL_ENTRYPOINT_ENV = "DEMAND_LOCAL_ENTRYPOINT" LOCAL_ENTRYPOINT_NAME = "run_existing_execution_local" MYSQL_ENTRYPOINT_ENV = "DEMAND_MYSQL_ENTRYPOINT" MYSQL_ENTRYPOINT_NAMES = {"run_existing_execution_mysql", "run_hive_gap_mysql"} def _resolve_run_platform(platform_type: Optional[str], demand_scope: Optional[dict[str, Any]]) -> str: scope = demand_scope if isinstance(demand_scope, dict) else {} raw_platform = platform_type or scope.get("platform") or scope.get("platform_type") or DEFAULT_DEMAND_PLATFORM return normalize_platform(raw_platform) def _is_local_json_mode() -> bool: return os.getenv("DEMAND_OUTPUT_MODE", "").strip().lower() == LOCAL_JSON_MODE def _is_mysql_demand_content_mode() -> bool: return os.getenv("DEMAND_OUTPUT_MODE", "").strip().lower() == MYSQL_DEMAND_CONTENT_MODE def _raise_db_write_disabled(action: str) -> None: raise RuntimeError( f"{action} 已在 PG Pattern V2 MVP 中禁用；" "请通过 run_existing_execution_local 写本地 JSON，或通过 " "run_existing_execution_mysql 只写测试 MySQL demand_content" ) 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 _require_mysql_entrypoint() -> None: if not _is_mysql_demand_content_mode(): return entrypoint = os.getenv(MYSQL_ENTRYPOINT_ENV, "").strip() if entrypoint not in MYSQL_ENTRYPOINT_NAMES: raise RuntimeError( "DEMAND_OUTPUT_MODE=mysql_demand_content 只能通过 " "examples.demand.run_existing_execution_mysql 或 " "examples.demand.run_hive_gap_mysql 入口运行" ) 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_WEIGHT_DATA_DIR", str(root / "intermediate" / "data")) 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_weight_data_base_dir() -> Path: redirected = os.getenv("DEMAND_WEIGHT_DATA_DIR") if redirected: return Path(redirected) return Path(__file__).parent / "data" 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): """Return the latest successful PG Pattern V2 execution_id. PG Pattern V2 executions are global snapshots rather than the old MySQL merge_leve2-scoped execution rows. Local MVP callers should pass an explicit execution_id; this fallback only keeps non-local paths from using MySQL. """ del cluster_name return query_latest_success_execution_id() def resolve_model(prompt: Any, run_config: Any) -> str: for env_key in ("DEMAND_LLM_MODEL", "ARK_MODEL", "OPEN_ROUTER_MODEL", "OPENROUTER_MODEL"): model_from_env = os.getenv(env_key) if model_from_env: return model_from_env.strip() 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 def extract_assistant_text(message: Any) -> str: if message.role != "assistant": return "" content = message.content if isinstance(content, str): return content if isinstance(content, dict): text = content.get("text", "") # 即使本轮包含工具调用，也打印模型给出的文本，便于观察每一步输出 if text: return text return "" def register_selected_tools(tool_names: list[str]) -> None: for tool_name in tool_names: module_path = CUSTOM_TOOL_MODULES.get(tool_name) if not module_path: raise ValueError(f"未配置工具模块映射: {tool_name}") importlib.import_module(module_path) def _enabled_tools_for_run(configured_tools: list[str]) -> list[str]: tools = configured_tools.copy() if _is_mysql_demand_content_mode(): allowed = { "think_and_plan", "get_category_tree", "get_frequent_itemsets", "get_itemset_detail", "get_post_elements", "search_elements", "get_element_category_chain", "get_category_detail", "search_categories", "get_category_elements", "get_category_co_occurrences", "get_element_co_occurrences", "get_weight_score_topn", "get_weight_score_by_name", "create_demand_item", "create_demand_items", } return [tool_name for tool_name in tools if tool_name in allowed] if _is_local_json_mode(): # 本地批量输出只需要 DemandItem JSON。长摘要会显著放大最后一轮上下文， # 且不参与下游 CFA 验证契约。 tools = [tool_name for tool_name in tools if tool_name != "write_execution_summary"] return tools def _join_element_names_to_name(element_names: object) -> str: """把 tool 入参 element_names 转成 demand_content.name（逗号分隔）。""" if element_names is None: return "" if isinstance(element_names, list): parts = [str(x).strip() for x in element_names if x is not None and str(x).strip()] return ",".join(parts) # 兼容异常数据：比如历史版本可能传了字符串 return str(element_names).strip() def _safe_truncate(s: object, max_len: int) -> str: if s is None: return "" s_str = str(s) if max_len and len(s_str) > max_len: return s_str[:max_len] return s_str def _load_name_score_map(execution_id: int) -> dict: """读取 data/{execution_id} 下所有 JSON 的「名字->score」（同名取最高分）。 兼容两类数据结构： - `*_元素.json`：字段 `name` 表示名字 - `*_分类.json`：字段 `category` 表示名字 """ data_dir = _get_weight_data_base_dir() / str(execution_id) if not data_dir.exists(): return {} score_map = {} for json_path in data_dir.glob("*.json"): try: with open(json_path, "r", encoding="utf-8") as f: payload = json.load(f) except Exception: continue if not isinstance(payload, list): continue for item in payload: if not isinstance(item, dict): continue # 元素数据以 name 为主；分类数据以 category 为主。 name = item.get("name") if not isinstance(name, str) or not name: name = item.get("category") score = item.get("score") if isinstance(name, str) and isinstance(score, (int, float)): prev = score_map.get(name) score_f = float(score) if prev is None or score_f > prev: score_map[name] = score_f return score_map def _ensure_pg_weight_score_files(execution_id: int) -> None: """Build local PG weight files when local JSON mode needs them.""" if not (_is_local_json_mode() or _is_mysql_demand_content_mode()): return from examples.demand.pg_weight_score_builder import build_pg_weight_score_files base_dir = _get_weight_data_base_dir() expected_dir = base_dir / str(execution_id) required_files = [ expected_dir / f"{dimension}_{level}.json" for dimension in ("实质", "形式", "意图") for level in ("元素", "分类") ] if all(path.exists() for path in required_files): return counts = build_pg_weight_score_files(execution_id=execution_id, base_dir=base_dir) log(f"[weight][pg] 生成 PG 权重 JSON，execution_id={execution_id}, files={counts}") def _avg_score_for_joined_name(name: str, score_map: dict) -> float: """按逗号拆分 name，分别取分后求平均。""" parts = [part.strip() for part in str(name).split(",") if part and part.strip()] if not parts: return 0.0 return sum(float(score_map.get(part, 0.0)) for part in parts) / len(parts) def _resolve_video_ids_by_name_and_execution_id(name: str, execution_id: int) -> list[str]: """按 name(逗号分隔) 与 execution_id 解析去重后的 post_id 列表。""" name_parts = [part.strip() for part in str(name).split(",") if part and part.strip()] if not name_parts: return [] 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 _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, demand_scope: Optional[dict[str, Any]] = None, ) -> 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, demand_scope=demand_scope or TopicBuildAgentContext.get_metadata("demand_scope", {}), merge_leve2=merge_level2, platform=TopicBuildAgentContext.get_metadata("platform"), ) 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 _create_demand_task( execution_id: int, name: Optional[str] = None, platform: Optional[str] = None, ) -> Optional[int]: """Disabled production demand_task writer retained for legacy imports.""" del execution_id, name, platform _raise_db_write_disabled("生产 _create_demand_task") def _finish_demand_task(task_id: Optional[int], status: int, task_log: str) -> None: """Disabled production demand_task updater retained for legacy imports.""" if not task_id: return del status, task_log _raise_db_write_disabled("生产 _finish_demand_task") 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, ) log(f"[task][local_json] 写入本地 demand_task，task_id={local_task_id}, execution_id={execution_id}") return local_task_id 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}") def write_demand_items_to_mysql( execution_id: int, merge_level2: str, *, trace_id: Optional[str] = None, task_id: Optional[int] = None, demand_scope: Optional[dict[str, Any]] = None, ) -> int: """Write DB-validated demand_content rows to the test MySQL sink only.""" if not _is_mysql_demand_content_mode(): _raise_db_write_disabled("生产 write_demand_items_to_mysql") from examples.demand.mysql_demand_content_sink import write_demand_content_rows 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, demand_scope=demand_scope, ) run_label = os.getenv("DEMAND_RUN_LABEL", "").strip() result = write_demand_content_rows(rows, run_label=run_label) log( "[mysql_demand_content] 写入测试 MySQL demand_content 完成，" f"demand_items={len(demand_items)}, inserted={result.inserted_count}, " f"rejected={len(rejected_items)}, skipped={result.skipped_count}" ) if rejected_items: log( "[mysql_demand_content] rejected_items=" + json.dumps(rejected_items, ensure_ascii=False, default=str) ) if result.inserted_ids: log(f"[mysql_demand_content] inserted_ids={result.inserted_ids}") return result.inserted_count def write_demand_items_to_local_json( execution_id: int, merge_level2: str, *, trace_id: Optional[str] = None, task_id: Optional[int] = None, demand_scope: Optional[dict[str, Any]] = 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, demand_scope=demand_scope, ) 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: """Disabled production Hive writer retained for legacy imports.""" del execution_id, merge_level2 _raise_db_write_disabled("生产 write_global_tree_demand_items_to_hive") 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, demand_scope: Optional[dict[str, Any]] = 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) if _is_local_json_mode() or _is_mysql_demand_content_mode(): _ensure_pg_weight_score_files(int(execution_id)) platform_type = _resolve_run_platform(platform_type, demand_scope) TopicBuildAgentContext.set_execution_id(execution_id) TopicBuildAgentContext.set_metadata("result_base_dir", str(_get_result_base_dir())) TopicBuildAgentContext.set_metadata("merge_leve2", merge_level2) TopicBuildAgentContext.set_metadata("platform", platform_type) resolved_demand_scope = dict(demand_scope or {}) resolved_demand_scope.setdefault("scope_source", "manual_cli") resolved_demand_scope.setdefault("merge_leve2", merge_level2) resolved_demand_scope["platform"] = _resolve_run_platform(platform_type, resolved_demand_scope) resolved_demand_scope.setdefault("pattern_execution_id", int(execution_id)) TopicBuildAgentContext.set_metadata("demand_scope", resolved_demand_scope) base_dir = Path(__file__).parent output_dir = _get_output_base_dir() output_dir.mkdir(parents=True, exist_ok=True) setup_logging(level=LOG_LEVEL, file=LOG_FILE) enabled_tools = _enabled_tools_for_run(ENABLED_TOOLS) register_selected_tools(enabled_tools) prompt_file = "demand_mysql.md" if _is_mysql_demand_content_mode() else "demand.md" prompt = SimplePrompt(base_dir / prompt_file) run_config = copy.deepcopy(RUN_CONFIG) model = resolve_model(prompt, run_config) run_config.model = model 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 # 禁用反思/总结经验相关流程（避免进入 reflection 侧分支） run_config.enable_research_flow = False run_config.goal_compression = "none" run_config.force_side_branch = None run_config.knowledge.enable_extraction = False run_config.knowledge.enable_completion_extraction = False run_config.knowledge.enable_injection = False run_config.trace_id = None initial_messages = prompt.build_messages(merge_level2=merge_level2, count=count) 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), skills_dir=None, debug=DEBUG, ) 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) try: 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) has_completed_trace_cost = True continue elif isinstance(item, Message): text = extract_assistant_text(item) if text: final_text = text log(f"[assistant] {text}") if not has_completed_trace_cost: total_tokens += int(getattr(item, "total_tokens", 0) or 0) total_cost += float(getattr(item, "cost", 0.0) or 0.0) if final_text: output_file = output_dir / f"{execution_id}" / "result.txt" output_file.parent.mkdir(parents=True, exist_ok=True) with open(output_file, "w", encoding="utf-8") as f: f.write(final_text) log(f"[cost] total_tokens={total_tokens}, total_cost=${total_cost:.6f}") # agent 执行完成后：全局树写 Hive，其他写 MySQL try: if str(merge_level2).strip() == "全局树": 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 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, demand_scope=resolved_demand_scope, ) 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, demand_scope=resolved_demand_scope, ) 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() if task_status != 2: task_status = 1 except Exception as e: if not task_log_text: # 如果异常发生在 build_log 内部，尽量回收已产生的日志 try: existing = locals().get("log_buffer") if existing is not None: task_log_text = existing.getvalue() # type: ignore[attr-defined] except Exception: pass if not task_log_text: task_log_text = f"[run] 执行异常: {e}" task_status = 2 raise finally: if task_log_text: try: with open(log_file_path, "w", encoding="utf-8") as f: f.write(task_log_text) except Exception: # 兜底：即使写文件失败，也要确保 MySQL 状态被更新 pass if _is_local_json_mode(): _finish_local_demand_task(task_id=task_id, status=task_status, task_log=task_log_text) elif not _is_mysql_demand_content_mode(): _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": 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( cluster_name: str, platform_type: str, count, execution_id: Optional[int] = None, task_id: Optional[int] = None, demand_scope: Optional[dict[str, Any]] = None, ) -> dict: if not (_is_local_json_mode() or _is_mysql_demand_content_mode()): _raise_db_write_disabled("非受控运行入口") 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 _is_mysql_demand_content_mode(): _require_mysql_entrypoint() if execution_id is None: raise ValueError("mysql_demand_content 模式必须传入已有 execution_id，不能触发 prepare 写库流程") if str(cluster_name).strip() == "全局树": raise ValueError("mysql_demand_content 模式只写普通 demand_content，不支持全局树 Hive 输出") platform_type = _resolve_run_platform(platform_type, demand_scope) if execution_id is None: execution_id = get_execution_id_by_merge_level2(cluster_name) if not execution_id: 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, ) run_result = await run_once( execution_id, cluster_name, count=count, task_id=task_id, platform_type=platform_type, demand_scope=demand_scope, ) 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__": asyncio.run(main('全局树', 'piaoquan', 50)) # write_demand_items_to_mysql(execution_id=8, merge_level2='贪污腐败')