feat: switch demand agent to pg pattern v2 and ark deepseek

.env

@@ -0,0 +1,129 @@
+# DemandAgentNew local environment
+# This file is ignored by git. Edit local secrets here; do not commit it.
+# Values were assembled from .env.template and 数据工程/.secrets/data-engineering.local.env.
+
+# LLM providers
+# Required by current examples/demand/config.py model through agent/llm/openrouter.py
+OPEN_ROUTER_API_KEY=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
+OPENROUTER_PROXY=
+HTTPS_PROXY=
+HTTP_PROXY=
+ALL_PROXY=
+QWEN_BASE_URL=
+QWEN_API_KEY=
+GEMINI_API_KEY=
+YESCODE_BASE_URL=
+YESCODE_API_KEY=
+
+# DemandAgent PG Pattern V2 source
+# DemandAgent-specific PG Pattern V2 DSN; takes precedence over PGVECTOR_DSN
+DEMAND_PATTERN_PG_DSN=postgresql://aiddit_aigc:%25a%26%26yqNxg%5EV1%24toJ%2AWOa%5E-b%5EX%3DQJ@gp-t4n72471pkmt4b9q7o-master.gpdbmaster.singapore.rds.aliyuncs.com:5432/open_aigc
+# Shared PG fallback used by pg_pattern_repository.py
+PGVECTOR_DSN=postgresql://aiddit_aigc:%25a%26%26yqNxg%5EV1%24toJ%2AWOa%5E-b%5EX%3DQJ@gp-t4n72471pkmt4b9q7o-master.gpdbmaster.singapore.rds.aliyuncs.com:5432/open_aigc
+DATA_ENGINEERING_OPEN_AIGC_PG_DSN=postgresql://aiddit_aigc:%25a%26%26yqNxg%5EV1%24toJ%2AWOa%5E-b%5EX%3DQJ@gp-t4n72471pkmt4b9q7o-master.gpdbmaster.singapore.rds.aliyuncs.com:5432/open_aigc
+
+# DemandAgent local-json runtime
+# run_existing_execution_local.py sets these dynamically per run; leave them empty here unless debugging.
+DEMAND_OUTPUT_MODE=
+DEMAND_LOCAL_ENTRYPOINT=
+DEMAND_LOCAL_OUTPUT_ROOT=
+DEMAND_RESULT_BASE_DIR=
+DEMAND_WEIGHT_DATA_DIR=
+DEMAND_TRACE_STORE_PATH=
+DEMAND_OUTPUT_BASE_DIR=
+DEMAND_LOCAL_TASK_ID=
+
+# Agent runtime toggles
+no_proxy=*
+NO_PROXY=*
+AGENT_DISABLE_BUILTIN_SKILLS=1
+AGENT_DISABLE_BUILTIN_TOOL_REGISTRATION=1
+AGENT_STRICT_TOOL_SELECTION=1
+AGENT_DISABLE_SIDE_BRANCHES=1
+AGENT_DEBUG=
+AGENT_PRICING_CONFIG=
+SKILLS_DIR=./skills
+
+# Knowledge/browser optional tools
+KNOWHUB_API=http://43.106.118.91:9999
+BROWSER_USE_API_KEY=
+
+# DemandAgent web/scheduler local defaults
+# Local default: keep scheduler off unless explicitly testing web_api scheduler
+DEMAND_SCHEDULER_ENABLED=0
+
+# ODPS / Hive mirrors
+ODPS_ACCESS_ID=LTAI9EBa0bd5PrDa
+ODPS_ACCESS_KEY=vAalxds7YxhfOA2yVv8GziCg3Y87v5
+ODPS_PROJECT=loghubods
+ODPS_ENDPOINT=http://service.odps.aliyun.com/api
+
+# Data-engineering DB references copied for local editing
+DEMAND_AGENT_SOURCE_REPO=/Users/samlee/Documents/works/demandagent
+CONTENT_FIND_SOURCE_REPO=/Users/samlee/Documents/works/ContentFindAgent
+DEMAND_PATTERN_DB_HOST=rm-t4nh1xx6o2a6vj8qu3o.mysql.singapore.rds.aliyuncs.com
+DEMAND_PATTERN_DB_PORT=3306
+DEMAND_PATTERN_DB_NAME=open_aigc_pattern
+DEMAND_PATTERN_DB_USER=content_rw
+DEMAND_PATTERN_DB_PASSWORD=bC1aH4bA1lB0
+DEMAND_OPEN_AIGC_DB_HOST=192.168.202.204
+DEMAND_OPEN_AIGC_DB_PORT=3306
+DEMAND_OPEN_AIGC_DB_NAME=open_aigc
+DEMAND_OPEN_AIGC_DB_USER=wx2016_longvideo
+DEMAND_OPEN_AIGC_DB_PASSWORD=wx2016_longvideoP@assword1234
+PATTERN_GLOBAL_DB_HOST=192.168.202.204
+PATTERN_GLOBAL_DB_PORT=3306
+PATTERN_GLOBAL_DB_NAME=open_aigc_pattern
+PATTERN_GLOBAL_DB_USER=wx2016_longvideo
+PATTERN_GLOBAL_DB_PASSWORD=wx2016_longvideoP@assword1234
+CONTENT_SUPPLY_DB_HOST=rm-t4nh1xx6o2a6vj8qu3o.mysql.singapore.rds.aliyuncs.com
+CONTENT_SUPPLY_DB_PORT=3306
+CONTENT_SUPPLY_DB_NAME=content-deconstruction-supply
+CONTENT_SUPPLY_DB_USER=content_rw
+CONTENT_SUPPLY_DB_PASSWORD=bC1aH4bA1lB0
+DATA_ENGINEERING_OPEN_AIGC_HOST=gp-t4n72471pkmt4b9q7o-master.gpdbmaster.singapore.rds.aliyuncs.com
+DATA_ENGINEERING_OPEN_AIGC_PORT=5432
+DATA_ENGINEERING_OPEN_AIGC_NAME=open_aigc
+DATA_ENGINEERING_OPEN_AIGC_USER=aiddit_aigc
+DATA_ENGINEERING_OPEN_AIGC_PASSWORD=%a&&yqNxg^V1\$toJ*WOa^-b^X=QJ
+PATTERN_V2_OPEN_AIGC_HOST=gp-t4n72471pkmt4b9q7o-master.gpdbmaster.singapore.rds.aliyuncs.com
+PATTERN_V2_OPEN_AIGC_PORT=5432
+PATTERN_V2_OPEN_AIGC_NAME=open_aigc
+PATTERN_V2_OPEN_AIGC_USER=aiddit_aigc
+PATTERN_V2_OPEN_AIGC_PASSWORD=%a&&yqNxg^V1\$toJ*WOa^-b^X=QJ
+CLUSTERING_OPEN_AIGC_HOST=gp-t4n72471pkmt4b9q7o-master.gpdbmaster.singapore.rds.aliyuncs.com
+CLUSTERING_OPEN_AIGC_PORT=5432
+CLUSTERING_OPEN_AIGC_NAME=open_aigc
+CLUSTERING_OPEN_AIGC_USER=aiddit_aigc
+CLUSTERING_OPEN_AIGC_PASSWORD=%a&&yqNxg^V1\$toJ*WOa^-b^X=QJ
+XUANTI_DECODE_DB_HOST=rm-t4nh1xx6o2a6vj8qu3o.mysql.singapore.rds.aliyuncs.com
+XUANTI_DECODE_DB_PORT=3306
+XUANTI_DECODE_DB_USER=content_rw
+XUANTI_DECODE_DB_PASSWORD=bC1aH4bA1lB0
+XUANTI_DECODE_DB_NAME=content-deconstruction-supply
+XUANTI_DECODE_DB_CHARSET=utf8mb4
+
+# Legacy MySQL helper aliases
+# examples/demand/mysql/db_config.py reads DB_INFO or DB_*/MYSQL_* if legacy MySQL helpers are used.
+DB_INFO=
+DB_HOST=
+DB_PORT=
+DB_USER=
+DB_PASSWORD=
+DB_NAME=
+DB_CHARSET=
+MYSQL_HOST=
+MYSQL_PORT=
+MYSQL_USER=
+MYSQL_PASSWORD=
+MYSQL_DATABASE=
+MYSQL_CHARSET=
+
+# Feishu optional built-in tools
+FEISHU_APP_ID=
+FEISHU_APP_SECRET=
+# Ark DeepSeek OpenAI-compatible LLM
+ARK_API_KEY=ark-6ca740e2-a719-4478-a8f8-bbb97985a784-059ce
+ARK_BASE_URL=https://ark.cn-beijing.volces.com/api/v3
+ARK_MODEL=deepseek-v4-flash-260425
+DEMAND_LLM_MODEL=deepseek-v4-flash-260425

.env.template

@@ -5,6 +5,10 @@ BROWSER_USE_API_KEY=
 QWEN_BASE_URL=
 QWEN_API_KEY=
 
-# sonnet模型
-OPEN_ROUTER_API_KEY=
+# LLM provider
+OPEN_ROUTER_API_KEY=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
 
+# PG Pattern V2 read-only source for DemandAgent evidence.
+# DEMAND_PATTERN_PG_DSN takes precedence; PGVECTOR_DSN is the shared fallback.
+DEMAND_PATTERN_PG_DSN=
+PGVECTOR_DSN=

agent/llm/openrouter.py

@@ -45,6 +45,9 @@ def _select_http_proxy_url() -> Optional[str]:
     - 环境变量存在且非空：使用该代理
     - 环境变量不存在：不传 proxy（让 httpx 按直连/系统配置处理）
     """
+    if os.getenv("no_proxy") == "*" or os.getenv("NO_PROXY") == "*":
+        return None
+
     # 优先级：专用 OPENROUTER_PROXY > 通用 HTTPS/HTTP/ALL 代理变量
     for key in (
         "OPENROUTER_PROXY",
@@ -61,6 +64,35 @@ def _select_http_proxy_url() -> Optional[str]:
     return None
 
 
+def _get_openai_compatible_api_key() -> str:
+    ark_api_key = os.getenv("ARK_API_KEY")
+    if ark_api_key and (os.getenv("ARK_BASE_URL") or os.getenv("ARK_MODEL")):
+        return ark_api_key
+
+    api_key = (
+        os.getenv("OPEN_ROUTER_API_KEY")
+        or os.getenv("OPENROUTER_API_KEY")
+        or ark_api_key
+    )
+    if not api_key:
+        raise ValueError("OPEN_ROUTER_API_KEY or ARK_API_KEY environment variable not set")
+    return api_key
+
+
+def _get_openai_compatible_base_url() -> str:
+    base_url = (
+        os.getenv("ARK_BASE_URL")
+        or os.getenv("OPEN_ROUTER_BASE_URL")
+        or os.getenv("OPENROUTER_BASE_URL")
+        or "https://openrouter.ai/api/v1"
+    )
+    return base_url.rstrip("/")
+
+
+def _is_openrouter_base_url(base_url: str) -> bool:
+    return "openrouter.ai" in base_url.lower()
+
+
 # ── OpenRouter Anthropic endpoint: model name mapping ──────────────────────
 # Local copy of yescode's model tables so this module is self-contained.
 _OR_MODEL_EXACT = {
@@ -646,17 +678,16 @@ async def openrouter_llm_call(
             "cost": float
         }
     """
-    api_key = os.getenv("OPEN_ROUTER_API_KEY")
-    if not api_key:
-        raise ValueError("OPEN_ROUTER_API_KEY environment variable not set")
+    api_key = _get_openai_compatible_api_key()
+    base_url = _get_openai_compatible_base_url()
 
-    # Claude 模型走 Anthropic 原生端点，其余走 OpenAI 兼容端点
+    # Claude 模型在 OpenRouter 域名下走 Anthropic 原生端点；
+    # Ark 等 OpenAI-compatible endpoint 始终走 /chat/completions。
     provider = _detect_provider_from_model(model)
-    if provider == "anthropic":
+    if provider == "anthropic" and _is_openrouter_base_url(base_url):
         logger.debug("[OpenRouter] Routing Claude model to Anthropic native endpoint")
         return await _openrouter_anthropic_call(messages, model, tools, api_key, **kwargs)
 
-    base_url = "https://openrouter.ai/api/v1"
     endpoint = f"{base_url}/chat/completions"
 
     # 跨 Provider 续跑时，重写不兼容的 tool_call_id

demand_output_improve_plan.md

@@ -6,6 +6,8 @@
 
 本次 DemandAgent 侧补齐的是上游 exact evidence 基础：它能让需求单携带真实 DB 校验过的 Pattern/itemset/category 证据，但最终链路还要求 ContentFindAgent 读取、继承并沉淀结构化 `source_evidence`，否则最终结果仍无法从 `post_id/aweme_id` exact 回到 Pattern 和分类树节点。
 
+PG Pattern V2 替换更新：正式证据源已经切到 PG `open_aigc.public` 主事实表，`evidence_pack.pattern_source_system` 固定为 `pg_pattern_v2`。MVP 只使用 `pattern_itemset.scope='topic'` 的分类 Pattern；`topic_element`、script、paragraph、group scope 不进入下游 exact 输出，因为它们不能保证每条证据都有 `category_id/category_path` 可回到分类树节点。
+
 改动范围判断：**不改 agent 框架层**，只改 `examples/demand` 这个业务领域实现。`agent/` 框架里的 runner、tool registry、trace store、LLM 调用链路都保持不动。本次即使涉及 `@tool` 参数、`trace_id` 透传，也只是修改 Demand 业务工具和业务输出，不修改通用 Agent 框架机制。
 
 ## 模块改动归类
@@ -17,7 +19,10 @@
 | `examples/demand/demand.md` | 改 | Prompt 要求 LLM 在创建需求时带上输出证据引用。 |
 | `examples/demand/run.py` | 改 | 当前真实写库入口集中在这里，需要加本地 JSON 输出模式。 |
 | `examples/demand/data_query_tools.py` | 改 | 不改 `dwd_multi_demand_pool_di` / `feature_point_data` 业务字段；把现有 Hive 输出拆成可复用的纯映射函数，local 模式只调用映射不写 Hive。 |
-| `examples/demand/db_manager.py` | 改 | 新增只读查询函数，补齐 `case_ids/category/element/itemset` 证据。 |
+| `examples/demand/db_manager.py` | 改 | 保留旧函数名作为 PG facade，底层改读 PG Pattern V2，只读补齐 `category/element/itemset/post` 证据。 |
+| 新增 `examples/demand/pg_pattern_repository.py` | 新增 | PG Pattern V2 只读 repository，统一读取 `pattern_mining_*`、`pattern_itemset*`。 |
+| 新增 `examples/demand/pattern_builds/pg_pattern_service.py` | 新增 | 保持旧工具返回结构，底层替换为 PG `scope='topic'` Pattern 查询。 |
+| 新增 `examples/demand/pg_weight_score_builder.py` | 新增 | 从 PG topic 元素/分类聚合本地权重 JSON，替代旧 prepare 产物。 |
 | 新增 `examples/demand/evidence_pack_builder.py` | 新增 | 专门构造并强校验 `ext_data.evidence_pack`；校验失败的 DemandItem 进入 reject，不进入 `demand_content`。 |
 | 新增 `examples/demand/local_output_sink.py` | 新增 | 统一本地 JSON sink，稳定输出 7 个本地文件；不进入通用 Agent 框架。 |
 | 新增 `examples/demand/run_existing_execution_local.py` | 新增 | MVP 测试入口：用已有 `execution_id` 只读 DB，禁止跑写库 prepare。 |
@@ -95,7 +100,7 @@
 
 - `examples/demand/evidence_pack_builder.py`
   - 新增 `evidence_pack` 构造逻辑。
-  - 从真实 DB 只读补齐 itemset/category/element/case 证据。
+  - 从 PG Pattern V2 真实 DB 只读补齐 itemset/category/element/post 证据。
   - 这是 DemandAgent 输出字段补齐逻辑。
 
 - `examples/demand/local_output_sink.py`
@@ -106,7 +111,7 @@
 
 - `examples/demand/run_existing_execution_local.py`
   - 新增只读测试入口。
-  - 只允许传已有 `execution_id`。
+  - 只允许传已有 PG `pattern_mining_execution.id`。
   - 禁止调用 `run_mining`，避免写 Pattern 中间表。
 
 ## Key Changes
@@ -164,23 +169,23 @@ record = {
     "type": "pattern",
     "video_ids": ["..."],
     "evidence_pack": {
-      "pattern_source_system": "mysql_topic_pattern",
-      "pattern_execution_id": 756,
-      "mining_config_id": 1301,
+      "pattern_source_system": "pg_pattern_v2",
+      "pattern_execution_id": 581,
+      "mining_config_id": 2081,
       "source_kind": "pattern_itemset",
       "case_id_type": "post_id",
-      "source_post_id": "55157577",
+      "source_post_id": "02454b8dab4c13c9fa21042bdc92d122",
       "category_bindings": [],
       "element_bindings": [],
-      "itemset_ids": [123],
+      "itemset_ids": [1607313],
       "itemset_items": [
         {
-          "itemset_id": 123,
-          "category_id": 456,
-          "category_path": "目的点>实质>祝福表达>祝福词句",
+          "itemset_id": 1607313,
+          "category_id": 123456,
+          "category_path": "/形式/...",
           "point_type": "关键点",
-          "dimension": "实质",
-          "element_name": "祝福词句"
+          "dimension": "形式",
+          "element_name": null
         }
       ],
       "support": 0.18,
@@ -237,20 +242,20 @@ test_output_data/{run_id}/
 
 | 字段 | 主要来源 | 真实性判断 | 是否可能受 LLM 影响 |
 | --- | --- | --- | --- |
-| `pattern_source_system` | 代码按当前读取链路设置，MVP 固定为 `mysql_topic_pattern` | 标记只能追到 DemandAgent 当前消费的 MySQL Pattern Tree；若要追 PG Pattern V2，必须另补 PG ID 或同步映射 | 否 |
-| `pattern_execution_id` | 当前运行参数 / `topic_pattern_execution.id` | DB 可验证事实 | 否 |
-| `mining_config_id` | `topic_pattern_itemset.mining_config_id` | 必须来自已校验 itemset | 否 |
+| `pattern_source_system` | 代码按当前读取链路设置，MVP 固定为 `pg_pattern_v2` | 标记证据来自 PG `open_aigc.public` Pattern V2 主事实 | 否 |
+| `pattern_execution_id` | 当前运行参数 / PG `pattern_mining_execution.id` | 必须存在且 `status='success'` | 否 |
+| `mining_config_id` | PG `pattern_itemset.mining_config_id` + `pattern_mining_config.id` | 必须来自同一 execution，且 config `scope='topic'` | 否 |
 | `case_id_type` | 代码按证据类型设置，Pattern case 链路固定为 `post_id` | 用于标准化 `source_post_id/matched_post_ids/aweme_id` 的 ID 语义 | 否 |
-| `source_post_id` | LLM 候选 + `topic_pattern_itemset.matched_post_ids` 校验 | 必须出现在已校验 itemset 的 `matched_post_ids` 中，否则拒绝该需求 | 候选阶段受影响，最终输出必须 DB 校验通过 |
+| `source_post_id` | LLM 候选 + PG `pattern_itemset_post.post_id` 校验 | 必须出现在已校验 itemset 的 `matched_post_ids` 中，否则拒绝该需求 | 候选阶段受影响，最终输出必须 DB 校验通过 |
 | `itemset_ids` | LLM 在 `evidence_refs` 中给出候选，代码按 `execution_id` 回查校验 | 只有属于当前 execution 且和需求证据匹配时才进入 evidence_pack；否则该需求拒绝 | 候选阶段受影响，最终输出必须 DB 校验通过 |
-| `itemset_items[]` | `topic_pattern_itemset_item` + `topic_pattern_category` | 必须来自已校验 itemset，且每个 `category_id` 能 join 到当前 execution 的分类节点 | 否 |
-| `support` / `absolute_support` | `topic_pattern_itemset` | DB 可验证事实 | 否 |
-| `matched_post_ids` | `topic_pattern_itemset.matched_post_ids` | DB 可验证事实；数量必须等于 `absolute_support` | 否 |
-| `category_bindings` | `topic_pattern_itemset_item` + `topic_pattern_category` | 必须来自已校验 itemset/category；查不到则拒绝该需求 | 否 |
-| `element_bindings` | `topic_pattern_element` | 必须能在当前 execution 下精确绑定到需求 seed 或 itemset item；只查到候选时拒绝该需求 | 否 |
+| `itemset_items[]` | PG `pattern_itemset_item` + `pattern_mining_category` | 必须来自已校验 `scope='topic'` itemset，且每个 `category_id/category_path` 能 join 到当前 execution 的分类节点 | 否 |
+| `support` / `absolute_support` | PG `pattern_itemset` | DB 可验证事实 | 否 |
+| `matched_post_ids` | PG `pattern_itemset_post` 聚合 | DB 可验证事实；`COUNT(DISTINCT post_id)` 必须等于 `absolute_support` | 否 |
+| `category_bindings` | PG `pattern_itemset_item` + `pattern_mining_category` | 必须来自已校验 itemset/category；查不到则拒绝该需求 | 否 |
+| `element_bindings` | PG `pattern_mining_element`，限定 `source_table='post_decode_topic_point_element'` | 必须能在当前 execution/source_post 下精确绑定到 itemset item；只查到候选时拒绝该需求 | 否 |
 | `video_ids` | 优先来自已校验 itemset 的 `matched_post_ids` | 必须来自已校验证据链；不再把宽松反查结果作为下游证据 | 否 |
 | `case_ids` | 已校验 itemset 的 `matched_post_ids` | Pattern 来源下固定是 `post_id` 语义，和 `case_id_type=post_id` 对齐 | 否 |
-| `decode_case_ids` | `workflow_decode_task_result.channel_content_id IN video_ids` | 可选补充字段；命中不到不影响 Pattern itemset 来源通过 | 否 |
+| `decode_case_ids` | PG-only MVP 不再查旧 MySQL decode 表，保留为空数组兼容字段 | 可选补充字段；命中不到不影响 Pattern itemset 来源通过 | 否 |
 | `seed_terms` | LLM 候选 + 代码校验后的 element/category/itemset 反推 | 最终输出必须能被已校验证据覆盖；否则拒绝该需求 | 候选阶段受影响，最终输出必须被证据覆盖 |
 | `source_kind` | LLM 候选明确声明；若缺失但已有 `itemset_ids` 或 `source_tool=get_itemset_detail`，DemandAgent 内部可安全补齐为 `pattern_itemset` | MVP 只支持 `pattern_itemset`；补齐后仍必须走 DB 强校验，不是 Pattern itemset 则拒绝 | 候选阶段受影响，最终输出必须 DB 校验通过 |
 | `trace_id` | 现有 Agent runner / trace store | 框架事实 | 否 |
@@ -279,24 +284,25 @@ test_output_data/{run_id}/
 
 5. 输入仍从真实 DB 拉
 
-MVP 入口只接受已有 `execution_id`：
+MVP 入口只接受已有 PG `pattern_mining_execution.id`：
 
 ```bash
 DEMAND_OUTPUT_MODE=local_json python -m examples.demand.run_existing_execution_local \
-  --execution-id 756 \
+  --execution-id 581 \
   --merge-level2 贪污腐败 \
   --platform-type piaoquan \
   --count 5 \
-  --run-id local_756_smoke
+  --run-id local_581_smoke
 ```
 
-local 模式只能通过 `examples.demand.run_existing_execution_local` 进入；`run.py` 会校验 `DEMAND_LOCAL_ENTRYPOINT=run_existing_execution_local`。该入口必须传已有 `execution_id`，禁止调用 `run_mining` 或各平台 prepare，避免写 `topic_pattern_*`。
+local 模式只能通过 `examples.demand.run_existing_execution_local` 进入；`run.py` 会校验 `DEMAND_LOCAL_ENTRYPOINT=run_existing_execution_local`。该入口必须传已有 PG execution，禁止调用 `run_mining` 或各平台 prepare，避免写任何 Pattern 中间表。
 
 本地入口会把三类运行产物统一重定向到 `examples/demand/test_output_data/{run_id}` 下。可传 `--run-id` 指定目录名；如果传 `--output-root`，也必须位于 `examples/demand/test_output_data/` 之内。
 
 ```text
 {output_root}/
   intermediate/result/
+  intermediate/data/
   .trace/
   output/
   run_manifest.json
@@ -308,16 +314,16 @@ local 模式只能通过 `examples.demand.run_existing_execution_local` 进入
   feature_point_data.json
 ```
 
-若未传 `--output-root`，默认写到 `examples/demand/test_output_data/execution_{execution_id}_{timestamp}/`。本 MVP 不刷新本地权重缓存；现有 `get_weight_score_*` 仍依赖 `examples/demand/data/{execution_id}` 已存在。
+若未传 `--output-root`，默认写到 `examples/demand/test_output_data/execution_{execution_id}_{timestamp}/`。local 模式会从 PG `pattern_mining_element` 聚合生成兼容的权重 JSON 到 `intermediate/data/{execution_id}/`，`get_weight_score_*` 读取该目录，不写 `examples/demand/data`。
 
 ## Test Plan
 
-- 写库屏障测试：本地模式下 monkeypatch `mysql_db.insert/update/insert_many` 和 `execute_odps_sql`，若被调用则测试失败。
+- 写库屏障测试：本地模式下 monkeypatch `mysql_db.insert/update/insert_many`、`execute_odps_sql` 和 PG DML/DDL，若被调用则测试失败；PG repository 只允许 `SELECT/SHOW/EXPLAIN`。
 - 输出结构测试：校验本地 JSON 文件存在；重点校验 `demand_content.ext_data.evidence_pack`，同时确认 `dwd_multi_demand_pool_di.json` / `feature_point_data.json` 没有新增业务字段。
 - evidence 测试：校验所有进入 `demand_content` 的 `evidence_pack.pattern_execution_id/source_kind/seed_terms/trace_id/source_certainty/validation_status` 必填，`validation_status` 必须为 `passed`，`source_certainty` 必须为 `db_validated`。
 - Pattern itemset evidence 测试：当 `source_kind=pattern_itemset` 时，`pattern_source_system/case_id_type/source_post_id/mining_config_id/itemset_ids/itemset_items/matched_post_ids/support/absolute_support` 必填，且 `source_post_id` 必须在 `matched_post_ids` 中，`len(matched_post_ids)` 必须等于 `absolute_support`。
 - 严格拒绝测试：构造 LLM 声明错误 itemset、跨 execution itemset、缺少 `matched_post_ids/source_post_id/itemset_items/mining_config_id`、DB 校验失败等样本，验证这些 DemandItem 不进入 `demand_content`，只进入 `rejected_demand_items.json`。
-- 回归测试：默认不设置 `DEMAND_OUTPUT_MODE` 时，旧生产路径仍保持原来的函数入口和字段映射。
+- 回归测试：不改 `agent/` 通用框架；DemandAgent pattern 工具名保持不变，但底层均走 PG adapter。
 - 多 subagent 验证：分别做 DB 血缘审计、输出契约审计、旧能力回归审计，交叉确认没有改坏原能力。
 
 ## Assumptions
@@ -325,6 +331,6 @@ local 模式只能通过 `examples.demand.run_existing_execution_local` 进入
 - MVP 不做真实 DB 写入，也不做 DB schema migration。
 - 本次只调整 DemandAgent 输出，不改 `agent/` 通用框架。
 - 本次下游证据链只以 `demand_content.ext_data.evidence_pack` 为载体；`dwd_multi_demand_pool_di` 和 `feature_point_data` 不承载 evidence_pack。
-- DemandAgent 当前消费的是 MySQL `open_aigc_pattern.topic_pattern_*` 兼容层，没有 `topic_pattern_itemset_post`；post 支撑关系通过 `topic_pattern_itemset.matched_post_ids` 表达。
-- 本 MVP 只保证追到 DemandAgent 当前消费的 MySQL Pattern Tree；若产品要求追到 PG Pattern V2 主事实，需要补 PG ID 或 PG -> MySQL 映射证明。
-- 对纯 Pattern itemset 来源，`case_ids` 按 `post_id` 语义直接等于已校验的 `matched_post_ids`，`decode_case_ids` 再按 `workflow_decode_task_result.channel_content_id IN video_ids` 只读补齐且允许为空。`source_post_id`、`matched_post_ids`、`itemset_ids`、`itemset_items[]`、`mining_config_id`、`support`、`absolute_support` 必须强校验通过。
+- DemandAgent 当前正式证据源是 PG `open_aigc.public` Pattern V2 主事实；旧 MySQL Pattern 兼容层只作为历史背景，不再作为 evidence 来源。
+- 本 MVP 只输出 PG `pattern_itemset.scope='topic'`；`topic_element`、script、paragraph、group scope 因无法保证分类树节点闭合，不进入 exact evidence。
+- 对纯 Pattern itemset 来源，`case_ids` 按 `post_id` 语义直接等于已校验的 `matched_post_ids`，`decode_case_ids` 保留为空数组兼容字段。`source_post_id`、`matched_post_ids`、`itemset_ids`、`itemset_items[]`、`mining_config_id`、`support`、`absolute_support` 必须强校验通过。

demandagent给下游的迭代需求.md

@@ -4,7 +4,7 @@
 
 新版 ContentFindAgent 不能只靠旧版反查机制来恢复 Pattern、Case 和分类树血缘。
 
-旧版能拿到一些候选线索，例如用需求词去查历史 Pattern 元素，再回查 Case 解构点。但这条链路不是精确血缘，也不能稳定闭合到唯一的 Pattern、itemset 或 Case。
+旧版能拿到一些候选线索，例如用需求词去查历史 Pattern 元素，再回查 Case 解构点。但这条链路不是精确血缘，也不能稳定闭合到唯一的 Pattern、itemset 或 Case。新版 DemandAgent 的正式证据源改为 PG `open_aigc.public` Pattern V2 主事实表。
 
 为了新版能从需求出发稳定跑通数据源、Query、判断、游走、资产沉淀和策略学习，DemandAgent 或上游证据层需要把生成需求时已经知道的证据一起传给下游。
 
@@ -14,8 +14,8 @@
 
 | 字段 | 作用 |
 | --- | --- |
-| `pattern_source_system` | 标记 Pattern 证据来自 DemandAgent 当前 MySQL 兼容层，还是 PG Pattern V2 主事实 |
-| `pattern_execution_id` | 标记需求来自哪次 Pattern 挖掘执行 |
+| `pattern_source_system` | 固定为 `pg_pattern_v2`，标记 Pattern 证据来自 PG Pattern V2 主事实 |
+| `pattern_execution_id` | 标记需求来自哪次 PG `pattern_mining_execution` |
 | `mining_config_id` | 标记来源 Pattern 使用的具体挖掘配置 |
 | `source_kind` | 标记需求来源是聚类结果、Pattern 频繁项集、直接 Case，还是其他来源 |
 | `case_id_type` | 标准化 `caseid` 的语义，例如本链路中应明确为 `post_id` |
@@ -35,6 +35,8 @@
 
 V1 不一定马上改复杂 DB schema。可以先在 `demand_content.ext_data` 里增加一个结构化 `evidence_pack`，先把证据传下来。
 
+MVP 边界：DemandAgent 只输出 PG `pattern_itemset.scope='topic'` 的分类 Pattern 证据。`topic_element`、script、paragraph、group scope 暂不输出，因为它们不能保证每条证据都有 `category_id/category_path` 回到分类树节点。
+
 ## 2. 新版下游到底需要什么
 
 新版 ContentFindAgent 不是只拿一个需求词去搜内容。它需要知道这个需求从哪里来、为什么成立、应该怎么生成 Query、判断结果应该绑定回哪棵分类树，以及最终资产入库前是否可追溯。
@@ -173,7 +175,7 @@ DemandAgent 自己在生成 `video_ids` 时，用的是 `execution_id + name/cat
 
 | 字段 | 字段意义 | 下游用途 | 为什么不能靠 CFA 自己补 | 缺失风险 |
 | --- | --- | --- | --- | --- |
-| `pattern_source_system` | Pattern 证据来源系统，MVP 固定为 `mysql_topic_pattern` | 判断反查目标是 DemandAgent MySQL Pattern Tree 还是 PG Pattern V2 | CFA 只看需求单无法知道证据来自哪套 Pattern 表 | 把 MySQL 兼容层误当 PG 主事实 |
+| `pattern_source_system` | Pattern 证据来源系统，MVP 固定为 `pg_pattern_v2` | 判断反查目标是 PG Pattern V2 主事实 | CFA 只看需求单无法知道证据来自哪套 Pattern 表 | 把历史兼容层或候选反查误当 PG 主事实 |
 | `pattern_execution_id` | 当前需求来源的 Pattern 执行 ID | 数据源详情、Pattern 回查、策略学习 | CFA 只能按词反查，无法知道当时 execution | 同名词跨 execution 混淆 |
 | `mining_config_id` | 当前 itemset 所属挖掘配置 ID | 区分同一 execution 下不同 Pattern 配置 | CFA 事后只靠 post 反查会得到多个配置下的候选 itemset | 无法确定具体 Pattern 配置 |
 | `source_kind` | 来源类型：聚类结果、Pattern 频繁项集、直接 Case 等 | 决定后续流程：是否进 Query，是否先游走或先判断 | 只看 `type` 不够，不能区分真实来源路径 | 不同来源走错流程 |
@@ -187,7 +189,7 @@ DemandAgent 自己在生成 `video_ids` 时，用的是 `execution_id + name/cat
 | `absolute_support` | 绝对支撑素材数 | 判断有多少素材支撑 | 需求池当前没有保存 | 不能区分单例和稳定模式 |
 | `matched_post_ids` | 支撑 itemset 的帖子 ID 列表 | 找 Case、看原始素材、做回扣判断 | `video_ids` 不等价于 itemset 的 matched posts | Case 来源不准，支撑证据不准 |
 | `video_ids` / `case_ids` | 可直接作为 Case seed 的素材 ID；Pattern 来源下固定是 `post_id` 语义 | 数据源 Case、Query 输入、判断证据 | CFA 可偶尔从 decode 表命中，但不稳定 | 如果语义不清，容易把 post_id 和 decode row id 混用 |
-| `decode_case_ids` | `workflow_decode_task_result.id` 的可选补充 | 有 decode 原文时可快速定位 decode 行 | 并非所有 Pattern matched post 都能命中 decode 表 | 不能把 decode 缺失误判成 Pattern 证据失败 |
+| `decode_case_ids` | 兼容补充字段，PG-only MVP 可为空数组 | 后续如需要再挂 decode 行 ID | 并非所有 Pattern matched post 都需要 decode row ID 才能闭合 | 不能把 decode 缺失误判成 Pattern 证据失败 |
 | `seed_terms` | 真正用于 Query 的下层特征或元素词 | Query 输入素材 | `name` 可能是分类、关系或宽泛表达，不一定适合作 Query | Query 输入不稳定 |
 | `source_evidence` | 下游最终结果继承的结构化证据 | 写入 `demand_find_content_result` 或 source edge artifact | DemandAgent 只能给需求证据，无法替下游结果落库 | 最终结果又断在 `process_trace` 文本 |
 | `source_certainty` | 证据可信度，进入下游时必须为 `db_validated` | 下游判断是否能 exact 回溯 | CFA 无法判断 LLM 声明是否已被 DB 校验 | 把候选证据误当真实证据 |
@@ -212,29 +214,31 @@ DemandAgent 自己在生成 `video_ids` 时，用的是 `execution_id + name/cat
 | `ext_data` | `reason/desc/type/video_ids` |
 | `dt` | 当天日期 |
 
-同一份文档也指出：`workflow_decode_task_result` 是辅助回填逻辑，不是主生成 loop 默认步骤；Agent 输出主要靠 Prompt 约束，没有后置校验器强制每条需求都有权重或支持度。
+历史文档曾把 `workflow_decode_task_result` 作为辅助回填逻辑；PG-only MVP 不再依赖它闭合 Pattern 证据。Agent 输出主要靠 Prompt 约束，最终是否进入需求池由 DemandAgent 代码按 PG DB 强校验决定。
 
 ### 6.2 DemandAgent 代码审计
 
-DemandAgent 代码里确实有 Pattern 分类、元素和 itemset 字段。
+DemandAgent 新版代码里直接读取 PG Pattern V2 的分类、元素、itemset 和支撑帖子字段。
 
 代码证据：
 
-- `TopicPatternCategory`：`/Users/samlee/Documents/works/DemandAgentNew/examples/demand/models.py:45`
-- `TopicPatternElement`：`/Users/samlee/Documents/works/DemandAgentNew/examples/demand/models.py:63`
-- `TopicPatternItemset`：`/Users/samlee/Documents/works/DemandAgentNew/examples/demand/models.py:93`
-- `TopicPatternItemsetItem`：`/Users/samlee/Documents/works/DemandAgentNew/examples/demand/models.py:109`
+- PG repository：`/Users/samlee/Documents/works/DemandAgentNew/examples/demand/pg_pattern_repository.py`
+- PG tool adapter：`/Users/samlee/Documents/works/DemandAgentNew/examples/demand/pattern_builds/pg_pattern_service.py`
+- Evidence 强校验：`/Users/samlee/Documents/works/DemandAgentNew/examples/demand/evidence_pack_builder.py`
 
 关键字段包括：
 
 | 表 | 关键字段 |
 | --- | --- |
-| `topic_pattern_category` | `execution_id`, `source_type`, `name`, `path`, `level`, `parent_id` |
-| `topic_pattern_element` | `execution_id`, `post_id`, `point_type`, `element_type`, `name`, `category_id`, `category_path` |
-| `topic_pattern_itemset` | `execution_id`, `combination_type`, `item_count`, `support`, `absolute_support`, `dimensions`, `matched_post_ids` |
-| `topic_pattern_itemset_item` | `itemset_id`, `point_type`, `dimension`, `category_id`, `category_path`, `element_name` |
+| `pattern_mining_execution` | `id`, `status`, `topic_itemset_count`, `post_count`, `category_count`, `element_count` |
+| `pattern_mining_config` | `id`, `execution_id`, `scope`, `params` |
+| `pattern_itemset` | `id`, `execution_id`, `mining_config_id`, `scope`, `support`, `absolute_support`, `dimensions` |
+| `pattern_itemset_post` | `execution_id`, `itemset_id`, `post_id` |
+| `pattern_itemset_item` | `itemset_id`, `point_type`, `dimension`, `category_id`, `category_path`, `element_name` |
+| `pattern_mining_category` | `id`, `execution_id`, `source_type`, `path`, `level`, `parent_id`, `classified_as` |
+| `pattern_mining_element` | `execution_id`, `post_id`, `source_table`, `point_type`, `element_type`, `name`, `category_id`, `category_path` |
 
-也就是说，DemandAgent 内部能接触到这些字段。问题不是上游完全没有，而是写给下游时没有保留下来。
+也就是说，PG 主事实表已经具备 DemandAgent 输出下游证据链所需字段。新版问题不是“上游没有”，而是 DemandAgent 必须把这些字段校验后写进 `demand_content.ext_data.evidence_pack`，并要求下游继续继承。
 
 ### 6.3 旧 ContentFindAgent 代码审计
 
@@ -280,7 +284,7 @@ DB 侧验证结论：
 | subagent | 验证范围 | 结论 |
 | --- | --- | --- |
 | DemandAgent 代码与文档侧 | DemandItem、`demand_content`、`dwd_multi_demand_pool_di`、Pattern 模型字段 | DemandAgent 当前下游字段很窄，大量 Pattern lineage 在写需求池时丢失 |
-| DB 只读侧 | `demand_content`、`workflow_decode_task_result`、`topic_pattern_*` 抽样闭合 | 只能做到候选回查，不能唯一闭合 Pattern / itemset / Case |
+| 历史 DB 只读侧 | `demand_content`、`workflow_decode_task_result`、旧 `topic_pattern_*` 抽样闭合 | 只能做到候选回查，不能唯一闭合 Pattern / itemset / Case；新版正式证据源改为 PG Pattern V2 |
 
 两个方向结论一致。
 
@@ -295,34 +299,34 @@ V1 建议先不要大改表结构，先扩展 `demand_content.ext_data`。
   "reason": "...",
   "desc": "...",
   "type": "...",
-  "video_ids": ["55157577"],
+  "video_ids": ["02454b8dab4c13c9fa21042bdc92d122"],
   "evidence_pack": {
-    "pattern_source_system": "mysql_topic_pattern",
+    "pattern_source_system": "pg_pattern_v2",
     "source_kind": "pattern_itemset",
     "case_id_type": "post_id",
-    "source_post_id": "55157577",
-    "pattern_execution_id": 1987,
-    "mining_config_id": 1301,
-    "itemset_ids": [377530],
+    "source_post_id": "02454b8dab4c13c9fa21042bdc92d122",
+    "pattern_execution_id": 581,
+    "mining_config_id": 2081,
+    "itemset_ids": [1607313],
     "itemset_items": [
       {
-        "itemset_id": 377530,
-        "category_id": 43925,
-        "category_path": "目的点>实质>祝福表达>祝福词句",
+        "itemset_id": 1607313,
+        "category_id": 123456,
+        "category_path": "/形式/...",
         "point_type": "关键点",
-        "dimension": "实质",
-        "element_name": "祝福词句"
+        "dimension": "形式",
+        "element_name": null
       }
     ],
     "category_bindings": [],
     "element_bindings": [],
-    "support": 0.18,
-    "absolute_support": 12,
-    "matched_post_ids": ["55157577"],
-    "video_ids": ["55157577"],
-    "case_ids": ["55157577"],
+    "support": 0.04539396315170521,
+    "absolute_support": 579,
+    "matched_post_ids": ["02454b8dab4c13c9fa21042bdc92d122"],
+    "video_ids": ["02454b8dab4c13c9fa21042bdc92d122"],
+    "case_ids": ["02454b8dab4c13c9fa21042bdc92d122"],
     "decode_case_ids": [],
-    "seed_terms": ["祝福词句"],
+    "seed_terms": ["从 PG itemset item / category path 推导的真实词"],
     "trace_id": "...",
     "demand_task_id": 123,
     "demand_content_id": 456,
@@ -338,10 +342,10 @@ V1 建议先不要大改表结构，先扩展 `demand_content.ext_data`。
 | --- | --- |
 | V1 必须 | `pattern_source_system`, `source_kind`, `case_id_type`, `source_post_id`, `pattern_execution_id`, `mining_config_id`, `itemset_ids`, `itemset_items[]`, `category_bindings`, `element_bindings`, `support`, `absolute_support`, `matched_post_ids`, `video_ids/case_ids`, `seed_terms`, `trace_id`, `source_certainty`, `validation_status` |
 | V1 可选补充 | `decode_case_ids`，用于保存能命中 `workflow_decode_task_result` 的 decode 行 ID |
-| V1.1 增强 | 独立证据表、PG Pattern V2 ID 映射、下游完整候选池判断日志 |
+| V1.1 增强 | 独立证据表、下游完整候选池判断日志、topic_element 后续分类树补齐方案 |
 | 后续再拆表 | 如果 `evidence_pack` 变大，再考虑独立证据表或关系表 |
 
-注意：如果来源不是 Pattern 频繁项集，而是聚类结果或直接 Case，`itemset_ids/support/matched_post_ids` 可以为空，但最终下发给下游的 `evidence_pack.source_kind` 必须明确，不能让下游猜。
+注意：本次 MVP 只下发 `source_kind=pattern_itemset` 且 PG `scope='topic'` 的需求；如果来源不是该类 Pattern 频繁项集，暂不进入 exact 输出。
 
 ### 7.1 DemandAgent 必须保证的真实性边界
 
@@ -353,13 +357,13 @@ V1 建议先不要大改表结构，先扩展 `demand_content.ext_data`。
 {
   "source_kind": "pattern_itemset",
   "source_tool": "get_itemset_detail",
-  "itemset_ids": [377530],
-  "category_ids": [43925],
-  "source_post_id": "55157577",
+  "itemset_ids": [1607313],
+  "category_ids": [123456],
+  "source_post_id": "02454b8dab4c13c9fa21042bdc92d122",
   "case_ids": {
-    "pattern_itemset": ["55157577"]
+    "pattern_itemset": ["02454b8dab4c13c9fa21042bdc92d122"]
   },
-  "seed_terms": ["祝福词句"]
+  "seed_terms": ["从 PG itemset items 中提取的真实词"]
 }
 ```
 
@@ -369,24 +373,24 @@ DemandAgent 代码必须按真实 DB 校验并补齐 `evidence_pack`。对于 `s
 
 | 校验项 | 必须满足 |
 | --- | --- |
-| `pattern_execution_id` | `topic_pattern_execution.id` 必须存在且 `status='success'` |
+| `pattern_execution_id` | PG `pattern_mining_execution.id` 必须存在且 `status='success'` |
 | `source_kind` | 最终 `evidence_pack` 必须是 `pattern_itemset`；LLM 漏填时可由 `itemset_ids/source_tool` 安全推断，不是 Pattern itemset 则拒绝 |
-| `itemset_ids` | 每个 itemset 都属于当前 `pattern_execution_id`，且能查到 `mining_config_id/support/absolute_support/matched_post_ids` |
-| `mining_config_id` | 必须来自 `topic_pattern_itemset.mining_config_id`，并能 join 到同一 execution 的 `topic_pattern_mining_config` |
-| `source_post_id` | 必须在对应 itemset 的 `matched_post_ids` 中 |
-| `matched_post_ids` | 必须非空，数量必须等于 `absolute_support` |
-| `itemset_items[]` | 必须能从 `topic_pattern_itemset_item` 查到，并能 join 到当前 execution 的 `topic_pattern_category` |
+| `itemset_ids` | 每个 itemset 都属于当前 PG execution，且 `pattern_itemset.scope='topic'`，能查到 `mining_config_id/support/absolute_support/matched_post_ids` |
+| `mining_config_id` | 必须来自 PG `pattern_itemset.mining_config_id`，并能 join 到同一 execution 且 `scope='topic'` 的 `pattern_mining_config` |
+| `source_post_id` | 必须在对应 itemset 的 `pattern_itemset_post.post_id` 聚合结果中 |
+| `matched_post_ids` | 必须由 PG `pattern_itemset_post` 聚合得到，非空，`COUNT(DISTINCT post_id)` 必须等于 `absolute_support` |
+| `itemset_items[]` | 必须能从 PG `pattern_itemset_item` 查到，并能 join 到当前 execution 的 `pattern_mining_category` |
 | `category_bindings` | `category_id` 必须属于当前 `pattern_execution_id`，不能跨 execution 借节点 |
-| `element_bindings` | 必须能用 `execution_id + source_post_id + category_id + point_type + dimension/element_type + element_name(如有)` 精确绑定到 `topic_pattern_element`，只查到候选词或其他 matched post 不算通过 |
+| `element_bindings` | 必须能用 `execution_id + source_post_id + category_id + point_type + dimension/element_type + element_name(如有)` 精确绑定到 PG `pattern_mining_element`，且限定 `source_table='post_decode_topic_point_element'` |
 | `source_certainty` / `validation_status` | 进入 `demand_content` 时只能是 `db_validated` / `passed` |
 
 任一必需字段查不到、跨 execution、post 不在 matched posts、category 不在当前 execution、source post 精确 element binding 不成立，当前 DemandItem 都失败，不写入 `demand_content`。失败项只进入本地 `rejected_demand_items.json`，Agent 继续生成下一条需求。
 
-对 `source_kind=pattern_itemset` 的需求，`case_ids` 按 `post_id` 语义直接等于已校验的 `matched_post_ids`，因为当前 MySQL Pattern Tree 的主支撑关系来自 `topic_pattern_itemset.matched_post_ids`；可为空的是 `decode_case_ids`。`source_post_id/itemset_ids/itemset_items/mining_config_id/matched_post_ids/support/absolute_support` 不可为空且必须 DB 校验通过。
+对 `source_kind=pattern_itemset` 的需求，`case_ids` 按 `post_id` 语义直接等于已校验的 `matched_post_ids`，因为 PG Pattern V2 的主支撑关系来自 `pattern_itemset_post`；可为空的是兼容字段 `decode_case_ids`。`source_post_id/itemset_ids/itemset_items/mining_config_id/matched_post_ids/support/absolute_support` 不可为空且必须 DB 校验通过。
 
 ### 7.2 本地 MVP 输出方式
 
-本次 MVP 不真实写 `demand_task`、`demand_content`、Hive 或 `topic_pattern_*`，只通过 `examples.demand.run_existing_execution_local` 读取已有 `execution_id` 并把输出写到本地：
+本次 MVP 不真实写 `demand_task`、`demand_content`、Hive、PG 或 MySQL，只通过 `examples.demand.run_existing_execution_local` 读取已有 PG `pattern_mining_execution.id` 并把输出写到本地：
 
 ```text
 examples/demand/test_output_data/{run_id}/
@@ -398,6 +402,7 @@ examples/demand/test_output_data/{run_id}/
   dwd_multi_demand_pool_di.json
   feature_point_data.json
   intermediate/result/
+  intermediate/data/
   .trace/
   output/
 ```

examples/demand/db_manager.py

@@ -1,586 +1,59 @@
-import json
-from datetime import date, datetime
-from typing import Any, Iterable, Mapping, Sequence
+"""DemandAgent PG Pattern V2 query facade.
 
-from sqlalchemy import bindparam, create_engine, and_, or_, desc, text
-from sqlalchemy.orm import sessionmaker, Session
+Historically this module opened the MySQL Pattern compatibility schema. The PG
+migration keeps the function names used by DemandAgent, but all
+Pattern evidence reads now go through `pg_pattern_repository` and are read-only.
+"""
 
+from __future__ import annotations
 
-class DatabaseManager:
-    """数据库管理类"""
-
-    # mysql+pymysql://<用户名>:<密码>@<主机地址>:<端口>/<数据库名>?charset=utf8mb4
-    def __init__(self):
-        connection_string = (
-            f"mysql+pymysql://content_rw:bC1aH4bA1lB0@rm-t4nh1xx6o2a6vj8qu3o.mysql.singapore.rds.aliyuncs.com:3306/open_aigc_pattern?charset=utf8mb4"
-        )
-        self.engine = create_engine(connection_string, pool_pre_ping=True, pool_recycle=3600)
-        self.SessionLocal = sessionmaker(bind=self.engine, autoflush=False, autocommit=False)
+from examples.demand.pg_pattern_repository import (
+    query_case_ids_by_post_ids,
+    query_category_level,
+    query_element_bindings_for_items,
+    query_execution_for_evidence,
+    query_itemset_evidence,
+    query_itemset_items_with_categories,
+    query_latest_success_execution_id,
+    query_video_ids_by_names,
+)
 
-    def get_session(self) -> Session:
-        """获取数据库会话"""
-        return self.SessionLocal()
 
+class DatabaseManager:
+    """Legacy guard.
 
-class DatabaseManager2:
-    """数据库管理类"""
+    PG Pattern V2 does not expose a SQLAlchemy MySQL session. Old prepare paths
+    must be migrated before production use; local MVP uses explicit PG
+    execution_id and never calls `get_session()`.
+    """
 
-    # mysql+pymysql://<用户名>:<密码>@<主机地址>:<端口>/<数据库名>?charset=utf8mb4
-    def __init__(self):
-        connection_string = (
-            f"mysql+pymysql://content_rw:bC1aH4bA1lB0@rm-t4nh1xx6o2a6vj8qu3o.mysql.singapore.rds.aliyuncs.com:3306/open_aigc_pattern?charset=utf8mb4"
+    def get_session(self):
+        raise RuntimeError(
+            "DatabaseManager MySQL session was removed; use PG Pattern V2 "
+            "repository functions instead"
         )
-        self.engine = create_engine(connection_string, pool_pre_ping=True, pool_recycle=3600)
-        self.SessionLocal = sessionmaker(bind=self.engine, autoflush=False, autocommit=False)
-
-    def get_session(self) -> Session:
-        """获取数据库会话"""
-        return self.SessionLocal()
-
-
-def query_video_ids_by_names(execution_id: int, names: Iterable[str]) -> list[str]:
-    """按 execution_id + 名称列表查询去重后的 post_id。"""
-    clean_names = [str(n).strip() for n in names if n is not None and str(n).strip()]
-    if not clean_names:
-        return []
-
-    manager = DatabaseManager()
-    session = manager.get_session()
-    video_ids: set[str] = set()
-    try:
-        for name in clean_names:
-            categories = session.execute(
-                text(
-                    """
-                    SELECT id
-                    FROM topic_pattern_category
-                    WHERE execution_id = :execution_id AND name = :name
-                    """
-                ),
-                {"execution_id": execution_id, "name": name},
-            ).fetchall()
-            category_ids = [row[0] for row in categories if row and row[0] is not None]
-
-            if category_ids:
-                elements = session.execute(
-                    text(
-                        """
-                        SELECT post_id
-                        FROM topic_pattern_element
-                        WHERE execution_id = :execution_id
-                          AND category_id IN :category_ids
-                        """
-                    ).bindparams(bindparam("category_ids", expanding=True)),
-                    {"execution_id": execution_id, "category_ids": category_ids},
-                ).fetchall()
-            else:
-                elements = session.execute(
-                    text(
-                        """
-                        SELECT post_id
-                        FROM topic_pattern_element
-                        WHERE execution_id = :execution_id AND name = :name
-                        """
-                    ),
-                    {"execution_id": execution_id, "name": name},
-                ).fetchall()
-
-            for row in elements:
-                post_id = row[0] if row else None
-                if post_id is not None and str(post_id).strip():
-                    video_ids.add(str(post_id).strip())
-    finally:
-        session.close()
-
-    return list(video_ids)
-
-def query_category_level(execution_id: int, name: str) -> int | None:
-    """按 execution_id + 分类名称查询 topic_pattern_category.level。"""
-    clean_name = str(name).strip() if name is not None else ""
-    if not clean_name:
-        return None
-
-    manager = DatabaseManager()
-    session = manager.get_session()
-    try:
-        row = session.execute(
-            text(
-                """
-                SELECT level
-                FROM topic_pattern_category
-                WHERE execution_id = :execution_id AND name = :name
-                ORDER BY id DESC
-                LIMIT 1
-                """
-            ),
-            {"execution_id": execution_id, "name": clean_name},
-        ).first()
-        if not row:
-            return None
-        level = row[0]
-        return int(level) if level is not None else None
-    finally:
-        session.close()
-
-
-def _serialize_db_value(value: Any) -> Any:
-    if isinstance(value, (datetime, date)):
-        return value.isoformat()
-    return value
-
-
-def _row_to_dict(row: Any) -> dict[str, Any]:
-    if row is None:
-        return {}
-    mapping = getattr(row, "_mapping", row)
-    return {key: _serialize_db_value(value) for key, value in dict(mapping).items()}
-
-
-def _jsonish_to_list(value: Any) -> list[Any]:
-    if value is None:
-        return []
-    if isinstance(value, list):
-        return value
-    if isinstance(value, (tuple, set)):
-        return list(value)
-    if isinstance(value, bytes):
-        value = value.decode("utf-8")
-    if isinstance(value, str):
-        raw = value.strip()
-        if not raw:
-            return []
-        try:
-            parsed = json.loads(raw)
-        except json.JSONDecodeError:
-            return [part.strip() for part in raw.split(",") if part.strip()]
-        if isinstance(parsed, list):
-            return parsed
-        if parsed is None:
-            return []
-        return [parsed]
-    return [value]
-
-
-def _to_post_id_list(value: Any) -> list[str]:
-    post_ids: list[str] = []
-    seen: set[str] = set()
-    for item in _jsonish_to_list(value):
-        post_id = str(item).strip() if item is not None else ""
-        if post_id and post_id not in seen:
-            seen.add(post_id)
-            post_ids.append(post_id)
-    return post_ids
-
-
-def _to_int_list(values: Any) -> list[int]:
-    result: list[int] = []
-    seen: set[int] = set()
-    for value in _jsonish_to_list(values):
-        if value is None or value == "":
-            continue
-        try:
-            int_value = int(value)
-        except (TypeError, ValueError):
-            continue
-        if int_value not in seen:
-            seen.add(int_value)
-            result.append(int_value)
-    return result
-
-
-def _chunked(values: Sequence[Any], size: int = 500) -> Iterable[list[Any]]:
-    for i in range(0, len(values), size):
-        yield list(values[i:i + size])
-
-
-def _quote_mysql_identifier(identifier: str) -> str:
-    return "`" + str(identifier).replace("`", "``") + "`"
-
-
-def _resolve_table_ref(session: Session, table_name: str) -> str | None:
-    row = session.execute(
-        text(
-            """
-            SELECT
-                table_schema AS resolved_table_schema,
-                table_name AS resolved_table_name
-            FROM information_schema.tables
-            WHERE table_name = :table_name
-            ORDER BY
-              CASE
-                WHEN table_schema = DATABASE() THEN 0
-                WHEN table_schema = 'content-deconstruction-supply' THEN 1
-                ELSE 2
-              END,
-              table_schema
-            LIMIT 1
-            """
-        ),
-        {"table_name": table_name},
-    ).mappings().first()
-    if not row:
-        return None
-    return (
-        f"{_quote_mysql_identifier(row['resolved_table_schema'])}."
-        f"{_quote_mysql_identifier(row['resolved_table_name'])}"
-    )
-
-
-def query_execution_for_evidence(execution_id: int) -> dict[str, Any] | None:
-    """只读查询 Pattern execution 基础信息，用于 evidence_pack 校验。"""
-    manager = DatabaseManager()
-    session = manager.get_session()
-    try:
-        row = session.execute(
-            text(
-                """
-                SELECT
-                    id,
-                    cluster_name,
-                    merge_leve2,
-                    platform,
-                    account_name,
-                    post_limit,
-                    min_absolute_support,
-                    classify_execution_id,
-                    mining_configs,
-                    post_count,
-                    itemset_count,
-                    status,
-                    error_message,
-                    start_time,
-                    end_time
-                FROM topic_pattern_execution
-                WHERE id = :execution_id
-                LIMIT 1
-                """
-            ),
-            {"execution_id": int(execution_id)},
-        ).mappings().first()
-        if not row:
-            return None
-        result = _row_to_dict(row)
-        result["mining_configs"] = _jsonish_to_list(result.get("mining_configs"))
-        return result
-    finally:
-        session.close()
-
-
-def query_itemset_evidence(execution_id: int, itemset_ids: Iterable[Any]) -> list[dict[str, Any]]:
-    """只读查询当前 execution 下的 itemset 事实。"""
-    clean_ids = _to_int_list(itemset_ids or [])
-    if not clean_ids:
-        return []
-
-    manager = DatabaseManager()
-    session = manager.get_session()
-    try:
-        rows = session.execute(
-            text(
-                """
-                SELECT
-                    i.id,
-                    i.execution_id,
-                    i.mining_config_id,
-                    cfg.execution_id AS mining_config_execution_id,
-                    cfg.dimension_mode AS mining_config_dimension_mode,
-                    cfg.target_depth AS mining_config_target_depth,
-                    i.combination_type,
-                    i.item_count,
-                    i.support,
-                    i.absolute_support,
-                    i.dimensions,
-                    i.is_cross_point,
-                    i.matched_post_ids
-                FROM topic_pattern_itemset i
-                JOIN topic_pattern_mining_config cfg
-                  ON cfg.id = i.mining_config_id
-                 AND cfg.execution_id = :execution_id
-                WHERE i.execution_id = :execution_id
-                  AND i.id IN :itemset_ids
-                ORDER BY i.id
-                """
-            ).bindparams(bindparam("itemset_ids", expanding=True)),
-            {
-                "execution_id": int(execution_id),
-                "itemset_ids": clean_ids,
-            },
-        ).mappings().all()
-
-        result: list[dict[str, Any]] = []
-        for row in rows:
-            itemset = _row_to_dict(row)
-            itemset["id"] = int(itemset["id"])
-            itemset["execution_id"] = int(itemset["execution_id"])
-            itemset["mining_config_id"] = (
-                int(itemset["mining_config_id"])
-                if itemset.get("mining_config_id") is not None
-                else None
-            )
-            itemset["mining_config_execution_id"] = (
-                int(itemset["mining_config_execution_id"])
-                if itemset.get("mining_config_execution_id") is not None
-                else None
-            )
-            itemset["item_count"] = (
-                int(itemset["item_count"])
-                if itemset.get("item_count") is not None
-                else None
-            )
-            itemset["absolute_support"] = (
-                int(itemset["absolute_support"])
-                if itemset.get("absolute_support") is not None
-                else None
-            )
-            itemset["support"] = (
-                float(itemset["support"])
-                if itemset.get("support") is not None
-                else None
-            )
-            itemset["is_cross_point"] = bool(itemset.get("is_cross_point"))
-            itemset["dimensions"] = _jsonish_to_list(itemset.get("dimensions"))
-            itemset["matched_post_ids"] = _to_post_id_list(itemset.get("matched_post_ids"))
-            result.append(itemset)
-        by_id = {itemset["id"]: itemset for itemset in result}
-        return [by_id[itemset_id] for itemset_id in clean_ids if itemset_id in by_id]
-    finally:
-        session.close()
-
-
-def query_itemset_items_with_categories(
-        execution_id: int,
-        itemset_ids: Iterable[Any],
-) -> list[dict[str, Any]]:
-    """只读查询 itemset item，并校验其 category 属于当前 execution。"""
-    clean_ids = _to_int_list(itemset_ids or [])
-    if not clean_ids:
-        return []
-
-    manager = DatabaseManager()
-    session = manager.get_session()
-    try:
-        rows = session.execute(
-            text(
-                """
-                SELECT
-                    i.id AS itemset_item_id,
-                    i.itemset_id,
-                    i.point_type,
-                    i.dimension,
-                    i.category_id,
-                    i.category_path,
-                    i.element_name,
-                    c.id AS bound_category_id,
-                    c.execution_id AS category_execution_id,
-                    c.source_type AS category_source_type,
-                    c.name AS category_name,
-                    c.path AS category_full_path,
-                    c.level AS category_level,
-                    c.parent_id AS category_parent_id
-                FROM topic_pattern_itemset_item i
-                JOIN topic_pattern_itemset s
-                  ON s.id = i.itemset_id
-                 AND s.execution_id = :execution_id
-                LEFT JOIN topic_pattern_category c
-                  ON c.id = i.category_id
-                 AND c.execution_id = :execution_id
-                WHERE i.itemset_id IN :itemset_ids
-                ORDER BY i.itemset_id, i.id
-                """
-            ).bindparams(bindparam("itemset_ids", expanding=True)),
-            {"execution_id": int(execution_id), "itemset_ids": clean_ids},
-        ).mappings().all()
-
-        result: list[dict[str, Any]] = []
-        for row in rows:
-            item = _row_to_dict(row)
-            for key in (
-                    "itemset_item_id",
-                    "itemset_id",
-                    "category_id",
-                    "bound_category_id",
-                    "category_execution_id",
-                    "category_level",
-                    "category_parent_id",
-            ):
-                if item.get(key) is not None:
-                    item[key] = int(item[key])
-            item["category_found"] = item.get("bound_category_id") is not None
-            result.append(item)
-        return result
-    finally:
-        session.close()
-
-
-def query_element_bindings_for_items(
-        execution_id: int,
-        itemset_items: Sequence[Mapping[str, Any]],
-        post_ids: Iterable[Any] | None = None,
-        limit_per_item: int = 200,
-) -> list[dict[str, Any]]:
-    """只读按 itemset item 精确绑定 topic_pattern_element。"""
-    clean_post_ids = _to_post_id_list(post_ids)
-    if not itemset_items:
-        return []
-
-    manager = DatabaseManager()
-    session = manager.get_session()
-    try:
-        bindings: list[dict[str, Any]] = []
-        for raw_item in itemset_items:
-            item = dict(raw_item)
-            category_id = item.get("bound_category_id") or item.get("category_id")
-            if category_id is None:
-                continue
-
-            clauses = [
-                "execution_id = :execution_id",
-                "category_id = :category_id",
-            ]
-            params: dict[str, Any] = {
-                "execution_id": int(execution_id),
-                "category_id": int(category_id),
-            }
-
-            if item.get("point_type"):
-                clauses.append("point_type = :point_type")
-                params["point_type"] = item["point_type"]
-            if item.get("dimension"):
-                clauses.append("element_type = :dimension")
-                params["dimension"] = item["dimension"]
-            if item.get("element_name"):
-                clauses.append("name = :element_name")
-                params["element_name"] = item["element_name"]
-            if clean_post_ids:
-                clauses.append("post_id IN :post_ids")
-                params["post_ids"] = clean_post_ids
-
-            where_sql = " AND ".join(clauses)
-            count_stmt = text(
-                f"""
-                SELECT
-                    COUNT(*) AS matched_element_count,
-                    COUNT(DISTINCT post_id) AS matched_post_count
-                FROM topic_pattern_element
-                WHERE {where_sql}
-                """
-            )
-            posts_stmt = text(
-                f"""
-                SELECT DISTINCT post_id
-                FROM topic_pattern_element
-                WHERE {where_sql}
-                ORDER BY post_id
-                LIMIT :limit_per_item
-                """
-            )
-            samples_stmt = text(
-                f"""
-                SELECT
-                    id,
-                    post_id,
-                    point_type,
-                    point_text,
-                    element_type,
-                    name,
-                    category_id,
-                    category_path
-                FROM topic_pattern_element
-                WHERE {where_sql}
-                ORDER BY post_id, id
-                LIMIT :sample_limit
-                """
-            )
-            if clean_post_ids:
-                count_stmt = count_stmt.bindparams(bindparam("post_ids", expanding=True))
-                posts_stmt = posts_stmt.bindparams(bindparam("post_ids", expanding=True))
-                samples_stmt = samples_stmt.bindparams(bindparam("post_ids", expanding=True))
-
-            count_row = session.execute(count_stmt, params).mappings().first() or {}
-            post_rows = session.execute(
-                posts_stmt,
-                {**params, "limit_per_item": int(limit_per_item)},
-            ).mappings().all()
-            sample_rows = session.execute(
-                samples_stmt,
-                {**params, "sample_limit": min(int(limit_per_item), 20)},
-            ).mappings().all()
-
-            bindings.append(
-                {
-                    "itemset_id": item.get("itemset_id"),
-                    "itemset_item_id": item.get("itemset_item_id"),
-                    "category_id": int(category_id),
-                    "point_type": item.get("point_type"),
-                    "dimension": item.get("dimension"),
-                    "element_name": item.get("element_name"),
-                    "matched_element_count": int(count_row.get("matched_element_count") or 0),
-                    "matched_post_count": int(count_row.get("matched_post_count") or 0),
-                    "matched_post_ids": _to_post_id_list([row["post_id"] for row in post_rows]),
-                    "sample_elements": [_row_to_dict(row) for row in sample_rows],
-                }
-            )
-        return bindings
-    finally:
-        session.close()
-
-
-def query_case_ids_by_post_ids(post_ids: Iterable[Any]) -> list[dict[str, Any]]:
-    """只读按 post_id 补齐 workflow_decode_task_result.id。"""
-    clean_post_ids = _to_post_id_list(post_ids)
-    if not clean_post_ids:
-        return []
-
-    manager = DatabaseManager()
-    session = manager.get_session()
-    try:
-        table_ref = _resolve_table_ref(session, "workflow_decode_task_result")
-        if not table_ref:
-            return []
-
-        result: list[dict[str, Any]] = []
-        for chunk in _chunked(clean_post_ids):
-            rows = session.execute(
-                text(
-                    f"""
-                    SELECT
-                        id AS case_id,
-                        channel_content_id AS post_id
-                    FROM {table_ref}
-                    WHERE channel_content_id IN :post_ids
-                    ORDER BY id
-                    """
-                ).bindparams(bindparam("post_ids", expanding=True)),
-                {"post_ids": chunk},
-            ).mappings().all()
-            for row in rows:
-                item = _row_to_dict(row)
-                if item.get("case_id") is not None:
-                    item["case_id"] = int(item["case_id"])
-                if item.get("post_id") is not None:
-                    item["post_id"] = str(item["post_id"]).strip()
-                result.append(item)
-        return result
-    finally:
-        session.close()
-
-db2 = DatabaseManager2()
-
-
-def exist_cluster_tree(merge_level2):
-    session = db2.get_session()
 
-    exec_row = session.execute(
-        text("""
-            SELECT id
-            FROM cluster_execution
-            WHERE name LIKE :name AND status = 2
-            ORDER BY create_time DESC
-            LIMIT 1
-        """),
-        {"name": f"{merge_level2}%"},
-    ).mappings().first()
 
-    if not exec_row:
-        return False
-    return True
+def exist_cluster_tree(merge_level2: str) -> bool:
+    """Return whether a successful PG Pattern V2 execution is available.
+
+    The old implementation checked MySQL cluster tree state by merge_level2.
+    The PG-only MVP consumes existing Pattern V2 executions instead, so this is
+    a conservative readiness check used only by legacy web guards.
+    """
+    del merge_level2
+    return query_latest_success_execution_id() is not None
+
+
+__all__ = [
+    "DatabaseManager",
+    "exist_cluster_tree",
+    "query_case_ids_by_post_ids",
+    "query_category_level",
+    "query_element_bindings_for_items",
+    "query_execution_for_evidence",
+    "query_itemset_evidence",
+    "query_itemset_items_with_categories",
+    "query_latest_success_execution_id",
+    "query_video_ids_by_names",
+]

examples/demand/demand.md

@@ -79,6 +79,7 @@ $system$
 约束：
 
 - **本次 MVP 最终可写入的 DemandItem 只允许 `source_kind="pattern_itemset"`**。不要把 `element` / `category` / `co_occurrence` 作为最终 `evidence_refs.source_kind`，这些只能用于探索，不能用于创建需求。
+- Pattern 证据源是 PG Pattern V2，最终只允许使用 `pattern_itemset.scope="topic"` 的分类 Pattern；`topic_element`、script、paragraph、group scope 都不能作为最终证据。
 - `source_tool` 必须填写 `get_itemset_detail`，因为最终证据必须来自项集详情。
 - `itemset_ids` 必须来自 `get_frequent_itemsets` 返回的真实项集 ID，并且在创建需求前必须用 `get_itemset_detail` 查询过。
 - `source_post_id` 必须从 `get_itemset_detail` 返回的 `post_ids` 中选择一个真实帖子 ID；没有明确帖子时不要创建该 DemandItem。

examples/demand/demand_pattern_tools.py

@@ -25,7 +25,7 @@ from typing import Any
 from agent import tool
 from examples.demand.demand_agent_context import TopicBuildAgentContext
 from examples.demand.log_capture import log
-from examples.demand.pattern_builds import pattern_service
+from examples.demand.pattern_builds import pg_pattern_service as pattern_service
 
 
 def _log_tool_input(tool_name: str, params: dict):
@@ -203,7 +203,8 @@ def get_itemset_detail(itemset_ids) -> str:
     if not itemset_ids:
         return _log_tool_output("get_itemset_detail", "错误: itemset_ids 为空或无法解析")
 
-    data = pattern_service.get_itemset_posts(itemset_ids)
+    execution_id = TopicBuildAgentContext.get_execution_id()
+    data = pattern_service.get_itemset_posts(itemset_ids, execution_id=execution_id)
     if not data:
         return _log_tool_output("get_itemset_detail", f"未找到 itemset_ids={itemset_ids} 的项集")
 
@@ -360,7 +361,7 @@ def get_category_detail(category_id: int) -> str:
     - 查看同级兄弟节点，发现平行的内容领域
 
     Args:
-        category_id: 分类节点ID（TopicPatternCategory.id）。
+        category_id: PG Pattern V2 `pattern_mining_category.id`。
 
     Returns:
         分类详情的JSON字符串，含 category、ancestors、children、siblings、elements。

examples/demand/evidence_pack_builder.py

@@ -18,7 +18,8 @@ from examples.demand.db_manager import (
 
 
 SOURCE_KIND_PATTERN_ITEMSET = "pattern_itemset"
-PATTERN_SOURCE_SYSTEM = "mysql_topic_pattern"
+PATTERN_SOURCE_SYSTEM = "pg_pattern_v2"
+PATTERN_ITEMSET_SCOPE = "topic"
 
 
 def build_evidence_pack(
@@ -315,6 +316,16 @@ def _validate_itemset_facts(itemsets: list[dict[str, Any]]) -> str | None:
                 f"itemset_id {itemset_id} mining_config_id {itemset.get('mining_config_id')} "
                 f"does not belong to execution_id {execution_id}"
             )
+        if _clean_str(itemset.get("scope")) != PATTERN_ITEMSET_SCOPE:
+            return (
+                f"itemset_id {itemset_id} scope={itemset.get('scope')} is not "
+                f"{PATTERN_ITEMSET_SCOPE}"
+            )
+        if _clean_str(itemset.get("mining_config_scope")) != PATTERN_ITEMSET_SCOPE:
+            return (
+                f"itemset_id {itemset_id} mining_config_id {itemset.get('mining_config_id')} "
+                f"scope={itemset.get('mining_config_scope')} is not {PATTERN_ITEMSET_SCOPE}"
+            )
         if itemset.get("support") is None:
             return f"itemset_id {itemset_id} missing support"
         if itemset.get("absolute_support") is None:
@@ -423,7 +434,7 @@ def _validate_element_bindings(
         if not binding:
             return f"itemset_item_id {item_id} has no element binding"
         if int(binding.get("matched_element_count") or 0) <= 0:
-            return f"itemset_item_id {item_id} has no exact topic_pattern_element binding"
+            return f"itemset_item_id {item_id} has no exact PG topic element binding"
         if int(binding.get("matched_post_count") or 0) <= 0:
             return f"itemset_item_id {item_id} has no matched post binding"
     return None

examples/demand/pattern_builds/pg_pattern_service.py

@@ -0,0 +1,461 @@
+"""PG Pattern V2 adapter for DemandAgent pattern tools.
+
+The public function names mirror the old MySQL pattern service, but every query
+reads PG `open_aigc.public` Pattern V2 facts and only exposes topic-scope
+itemsets as exact evidence candidates.
+"""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from typing import Any, Iterable
+
+from examples.demand import pg_pattern_repository as repo
+
+
+def _clean_ints(values: Iterable[Any] | None) -> list[int]:
+    result: list[int] = []
+    seen: set[int] = set()
+    for raw in values or []:
+        try:
+            value = int(raw)
+        except (TypeError, ValueError):
+            continue
+        if value not in seen:
+            seen.add(value)
+            result.append(value)
+    return result
+
+
+def _last_path_part(path: Any) -> str:
+    text = str(path or "").strip()
+    if not text:
+        return ""
+    return text.replace(">", "/").split("/")[-1].strip()
+
+
+def _group_point_types(rows: list[dict[str, Any]]) -> list[str]:
+    result: list[str] = []
+    seen: set[str] = set()
+    for row in rows:
+        value = str(row.get("point_type") or "").strip()
+        if value and value not in seen:
+            seen.add(value)
+            result.append(value)
+    return result
+
+
+def _format_item(item: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "id": item.get("itemset_item_id"),
+        "itemset_id": item.get("itemset_id"),
+        "layer": item.get("layer"),
+        "point_type": item.get("point_type"),
+        "dimension": item.get("dimension"),
+        "category_id": item.get("category_id"),
+        "category_path": item.get("category_path") or item.get("category_full_path"),
+        "category_name": item.get("category_name") or _last_path_part(item.get("category_path")),
+        "element_name": item.get("element_name"),
+        "post_count": item.get("post_count"),
+    }
+
+
+def get_category_tree_compact(execution_id: int, source_type: str | None = None) -> str:
+    categories = repo.query_categories(execution_id, source_type=source_type)
+    if not categories:
+        return f"未找到 PG Pattern V2 分类树，execution_id={execution_id}, source_type={source_type}"
+
+    lines = [
+        f"PG Pattern V2 分类树快照 execution_id={execution_id}",
+        "说明：本 DemandAgent MVP 只使用 scope=topic 的分类 Pattern；topic_element 不进入最终证据。",
+    ]
+    for cat in categories:
+        level = int(cat.get("level") or 0)
+        indent = "  " * max(level - 1, 0)
+        lines.append(
+            f"{indent}- [{cat.get('id')}] {cat.get('name')} "
+            f"(type={cat.get('source_type')}, level={level}, elements={cat.get('element_count') or 0}, "
+            f"path={cat.get('path')})"
+        )
+    return "\n".join(lines)
+
+
+def search_top_itemsets(
+        execution_id: int,
+        top_n: int = 20,
+        category_ids: list | None = None,
+        dimension_mode: str | None = None,
+        min_support: int | None = None,
+        min_item_count: int | None = None,
+        max_item_count: int | None = None,
+        sort_by: str = "absolute_support",
+        account_name=None,
+        merge_leve2=None,
+) -> dict[str, Any]:
+    del account_name, merge_leve2  # PG Pattern V2 MVP does not filter itemsets by post metadata here.
+    rows = repo.query_topic_itemsets(
+        execution_id=execution_id,
+        category_ids=category_ids,
+        dimension_mode=dimension_mode,
+        min_support=min_support,
+        min_item_count=min_item_count,
+        max_item_count=max_item_count,
+        sort_by=sort_by,
+        limit=max(int(top_n or 20) * 10, int(top_n or 20)),
+    )
+    itemset_ids = [int(row["id"]) for row in rows]
+    items_by_itemset: dict[int, list[dict[str, Any]]] = defaultdict(list)
+    if itemset_ids:
+        for item in repo.query_itemset_items_with_categories(execution_id, itemset_ids):
+            items_by_itemset[int(item["itemset_id"])].append(_format_item(item))
+
+    groups: dict[str, dict[str, Any]] = {}
+    for row in rows:
+        dimension_key = row.get("dimension_mode") or row.get("mining_config_scope") or "topic"
+        target_depth = row.get("target_depth") or ""
+        key = f"{dimension_key}/{target_depth}"
+        group = groups.setdefault(
+            key,
+            {
+                "dimension_mode": dimension_key,
+                "target_depth": target_depth,
+                "total": 0,
+                "itemsets": [],
+            },
+        )
+        if len(group["itemsets"]) >= int(top_n or 20):
+            continue
+        group["total"] += 1
+        group["itemsets"].append(
+            {
+                "id": row.get("id"),
+                "scope": row.get("scope"),
+                "mining_config_id": row.get("mining_config_id"),
+                "dimension_mode": dimension_key,
+                "target_depth": target_depth,
+                "combination_type": row.get("combination_type"),
+                "item_count": row.get("item_count"),
+                "support": row.get("support"),
+                "absolute_support": row.get("absolute_support"),
+                "dimensions": row.get("dimensions") or [],
+                "items": items_by_itemset.get(int(row["id"]), []),
+            }
+        )
+
+    return {
+        "pattern_source_system": repo.PG_PATTERN_SOURCE_SYSTEM,
+        "scope": repo.TOPIC_SCOPE,
+        "total": len(rows),
+        "showing": sum(len(group["itemsets"]) for group in groups.values()),
+        "groups": groups,
+    }
+
+
+def get_itemset_posts(itemset_ids: list[int], execution_id: int | None = None) -> list[dict[str, Any]]:
+    clean_ids = _clean_ints(itemset_ids)
+    if not clean_ids:
+        return []
+    if execution_id is None:
+        raise ValueError("PG Pattern V2 get_itemset_posts requires execution_id")
+
+    itemsets = repo.query_itemset_evidence(execution_id=execution_id, itemset_ids=clean_ids)
+    itemset_items = repo.query_itemset_items_with_categories(execution_id=execution_id, itemset_ids=clean_ids)
+    items_by_itemset: dict[int, list[dict[str, Any]]] = defaultdict(list)
+    for item in itemset_items:
+        items_by_itemset[int(item["itemset_id"])].append(_format_item(item))
+
+    details: list[dict[str, Any]] = []
+    for itemset in itemsets:
+        details.append(
+            {
+                "id": itemset.get("id"),
+                "execution_id": itemset.get("execution_id"),
+                "scope": itemset.get("scope"),
+                "mining_config_id": itemset.get("mining_config_id"),
+                "dimension_mode": itemset.get("mining_config_dimension_mode"),
+                "target_depth": itemset.get("mining_config_target_depth"),
+                "combination_type": itemset.get("combination_type"),
+                "item_count": itemset.get("item_count"),
+                "support": itemset.get("support"),
+                "absolute_support": itemset.get("absolute_support"),
+                "dimensions": itemset.get("dimensions") or [],
+                "items": items_by_itemset.get(int(itemset["id"]), []),
+                "post_ids": itemset.get("matched_post_ids") or [],
+                "matched_post_ids": itemset.get("matched_post_ids") or [],
+            }
+        )
+    return details
+
+
+def get_post_elements(execution_id: int, post_ids: list) -> dict[str, Any]:
+    rows = repo.query_elements(execution_id, post_ids=post_ids, limit=5000)
+    result: dict[str, Any] = {}
+    grouped: dict[tuple[str, str, str], dict[str, Any]] = {}
+    for row in rows:
+        post_id = str(row.get("post_id") or "")
+        point_type = str(row.get("point_type") or "未分点")
+        point_text = str(row.get("point_text") or "")
+        key = (post_id, point_type, point_text)
+        point = grouped.setdefault(
+            key,
+            {
+                "point_text": point_text,
+                "elements": {"实质": [], "形式": [], "意图": []},
+            },
+        )
+        element_type = str(row.get("element_type") or "其他")
+        point["elements"].setdefault(element_type, []).append(
+            {
+                "id": row.get("id"),
+                "source_element_id": row.get("source_element_id"),
+                "name": row.get("name"),
+                "description": row.get("description"),
+                "category_id": row.get("category_id"),
+                "category_path": row.get("category_path"),
+            }
+        )
+    for (post_id, point_type, _), point in grouped.items():
+        result.setdefault(post_id, {}).setdefault(point_type, []).append(point)
+    return result
+
+
+def search_elements(
+        execution_id: int,
+        keyword: str,
+        element_type: str | None = None,
+        limit: int = 50,
+        account_name=None,
+        merge_leve2=None,
+) -> list[dict[str, Any]]:
+    del account_name, merge_leve2
+    rows = repo.query_elements(execution_id, keyword=keyword, element_type=element_type, limit=5000)
+    grouped: dict[tuple[str, str, int | None, str | None], list[dict[str, Any]]] = defaultdict(list)
+    for row in rows:
+        key = (
+            str(row.get("name") or ""),
+            str(row.get("element_type") or ""),
+            row.get("category_id"),
+            row.get("category_path"),
+        )
+        grouped[key].append(row)
+
+    items: list[dict[str, Any]] = []
+    for (name, etype, category_id, category_path), group_rows in grouped.items():
+        if not name:
+            continue
+        posts = {str(row.get("post_id")) for row in group_rows if row.get("post_id") is not None}
+        items.append(
+            {
+                "name": name,
+                "element_type": etype,
+                "category_id": category_id,
+                "category_path": category_path,
+                "point_types": _group_point_types(group_rows),
+                "occurrence_count": len(group_rows),
+                "post_count": len(posts),
+            }
+        )
+    items.sort(key=lambda item: (item["post_count"], item["occurrence_count"]), reverse=True)
+    return items[: int(limit or 50)]
+
+
+def get_element_category_chain(
+        execution_id: int,
+        element_name: str,
+        element_type: str | None = None,
+) -> list[dict[str, Any]]:
+    rows = repo.query_elements(execution_id, keyword=element_name, element_type=element_type, limit=2000)
+    exact_rows = [row for row in rows if str(row.get("name") or "") == str(element_name)]
+    category_ids = _clean_ints(row.get("category_id") for row in exact_rows if row.get("category_id"))
+    categories = {int(row["id"]): row for row in repo.query_categories(execution_id, category_ids=category_ids)}
+    results: list[dict[str, Any]] = []
+    seen: set[int] = set()
+    for row in exact_rows:
+        category_id = row.get("category_id")
+        if category_id is None or int(category_id) in seen:
+            continue
+        seen.add(int(category_id))
+        category = categories.get(int(category_id), {})
+        results.append(
+            {
+                "category_id": category_id,
+                "category_path": row.get("category_path") or category.get("path"),
+                "element_type": row.get("element_type"),
+                "point_types": _group_point_types([r for r in exact_rows if r.get("category_id") == category_id]),
+                "ancestors": _ancestors_from_path(category.get("path") or row.get("category_path")),
+            }
+        )
+    return results
+
+
+def _ancestors_from_path(path: Any) -> list[dict[str, Any]]:
+    parts = [part.strip() for part in str(path or "").replace(">", "/").split("/") if part.strip()]
+    return [{"name": part, "level": index + 1, "path": "/".join(parts[: index + 1])} for index, part in enumerate(parts)]
+
+
+def get_category_detail_with_context(execution_id: int, category_id: int) -> dict[str, Any] | None:
+    categories = repo.query_categories(execution_id, category_ids=[category_id])
+    if not categories:
+        return None
+    category = categories[0]
+    all_categories = repo.query_categories(execution_id)
+    children = [cat for cat in all_categories if cat.get("parent_id") == int(category_id)]
+    siblings = [
+        cat for cat in all_categories
+        if cat.get("parent_id") == category.get("parent_id") and cat.get("id") != category.get("id")
+    ][:50]
+    elements = get_category_elements(category_id, execution_id=execution_id)
+    return {
+        "category": category,
+        "ancestors": _ancestors_from_path(category.get("path")),
+        "children": children[:100],
+        "siblings": siblings,
+        "elements": elements[:100],
+    }
+
+
+def search_categories(execution_id: int, keyword: str, source_type: str | None = None) -> list[dict[str, Any]]:
+    categories = repo.query_categories(execution_id, source_type=source_type, keyword=keyword, limit=100)
+    if not categories:
+        return []
+    category_ids = _clean_ints(cat.get("id") for cat in categories)
+    element_rows = repo.query_elements(execution_id, category_ids=category_ids, limit=5000)
+    elements_by_category: dict[int, list[dict[str, Any]]] = defaultdict(list)
+    for row in element_rows:
+        if row.get("category_id") is not None:
+            elements_by_category[int(row["category_id"])].append(row)
+    result: list[dict[str, Any]] = []
+    for cat in categories:
+        rows = elements_by_category.get(int(cat["id"]), [])
+        result.append(
+            {
+                **cat,
+                "point_types": _group_point_types(rows),
+                "post_count": len({str(row.get("post_id")) for row in rows if row.get("post_id") is not None}),
+            }
+        )
+    return result
+
+
+def get_category_elements(
+        category_id: int,
+        execution_id: int,
+        account_name=None,
+        merge_leve2=None,
+) -> list[dict[str, Any]]:
+    del account_name, merge_leve2
+    rows = repo.query_elements(execution_id, category_ids=[category_id], limit=5000)
+    grouped: dict[tuple[str, str], list[dict[str, Any]]] = defaultdict(list)
+    for row in rows:
+        name = str(row.get("name") or "").strip()
+        element_type = str(row.get("element_type") or "").strip()
+        if name:
+            grouped[(name, element_type)].append(row)
+    result: list[dict[str, Any]] = []
+    for (name, element_type), group_rows in grouped.items():
+        result.append(
+            {
+                "name": name,
+                "element_type": element_type,
+                "point_types": _group_point_types(group_rows),
+                "occurrence_count": len(group_rows),
+                "post_count": len({str(row.get("post_id")) for row in group_rows if row.get("post_id") is not None}),
+            }
+        )
+    result.sort(key=lambda item: (item["post_count"], item["occurrence_count"]), reverse=True)
+    return result
+
+
+def get_category_co_occurrences(
+        execution_id: int,
+        category_ids: list,
+        top_n: int = 30,
+        account_name=None,
+        merge_leve2=None,
+) -> dict[str, Any]:
+    del account_name, merge_leve2
+    clean_ids = _clean_ints(category_ids)
+    if not clean_ids:
+        return {"matched_post_count": 0, "co_categories": []}
+    rows = repo.query_elements(execution_id, category_ids=clean_ids, limit=10000)
+    posts_by_category: dict[int, set[str]] = defaultdict(set)
+    for row in rows:
+        if row.get("category_id") is not None and row.get("post_id") is not None:
+            posts_by_category[int(row["category_id"])].add(str(row["post_id"]))
+    matched_posts: set[str] | None = None
+    for category_id in clean_ids:
+        current = posts_by_category.get(category_id, set())
+        matched_posts = current if matched_posts is None else matched_posts & current
+    matched_posts = matched_posts or set()
+    co_rows = repo.query_elements(execution_id, post_ids=list(matched_posts)[:5000], limit=20000)
+    counts: dict[tuple[int, str, str], set[str]] = defaultdict(set)
+    for row in co_rows:
+        cat_id = row.get("category_id")
+        if cat_id is None or int(cat_id) in clean_ids:
+            continue
+        counts[(int(cat_id), str(row.get("category_path") or ""), str(row.get("element_type") or ""))].add(
+            str(row.get("post_id"))
+        )
+    ranked = sorted(counts.items(), key=lambda item: len(item[1]), reverse=True)
+    return {
+        "matched_post_count": len(matched_posts),
+        "matched_post_ids": sorted(matched_posts)[:100],
+        "co_categories": [
+            {
+                "category_id": key[0],
+                "category_path": key[1],
+                "element_type": key[2],
+                "post_count": len(posts),
+            }
+            for key, posts in ranked[: int(top_n or 30)]
+        ],
+    }
+
+
+def get_element_co_occurrences(
+        execution_id: int,
+        element_names: list,
+        top_n: int = 30,
+        account_name=None,
+        merge_leve2=None,
+) -> dict[str, Any]:
+    del account_name, merge_leve2
+    names = [str(name).strip() for name in element_names or [] if str(name).strip()]
+    if not names:
+        return {"matched_post_count": 0, "co_elements": []}
+    posts_by_name: dict[str, set[str]] = {}
+    for name in names:
+        rows = repo.query_elements(execution_id, keyword=name, limit=10000)
+        posts_by_name[name] = {
+            str(row.get("post_id"))
+            for row in rows
+            if str(row.get("name") or "") == name and row.get("post_id") is not None
+        }
+    matched_posts: set[str] | None = None
+    for posts in posts_by_name.values():
+        matched_posts = posts if matched_posts is None else matched_posts & posts
+    matched_posts = matched_posts or set()
+    co_rows = repo.query_elements(execution_id, post_ids=list(matched_posts)[:5000], limit=20000)
+    grouped: dict[tuple[str, str, int | None, str | None], list[dict[str, Any]]] = defaultdict(list)
+    for row in co_rows:
+        name = str(row.get("name") or "")
+        if not name or name in names:
+            continue
+        grouped[(name, str(row.get("element_type") or ""), row.get("category_id"), row.get("category_path"))].append(row)
+    ranked = sorted(grouped.items(), key=lambda item: len({r.get("post_id") for r in item[1]}), reverse=True)
+    return {
+        "matched_post_count": len(matched_posts),
+        "matched_post_ids": sorted(matched_posts)[:100],
+        "co_elements": [
+            {
+                "name": key[0],
+                "element_type": key[1],
+                "category_id": key[2],
+                "category_path": key[3],
+                "point_types": _group_point_types(rows),
+                "occurrence_count": len(rows),
+                "post_count": len({str(row.get("post_id")) for row in rows if row.get("post_id") is not None}),
+            }
+            for key, rows in ranked[: int(top_n or 30)]
+        ],
+    }

examples/demand/pg_pattern_repository.py

@@ -0,0 +1,804 @@
+"""Read-only PostgreSQL Pattern V2 repository for DemandAgent.
+
+This module is the DemandAgent boundary to PG `open_aigc.public` Pattern V2
+facts. It intentionally exposes the old Demand-side query shapes while reading
+only PG main-truth tables:
+
+- pattern_mining_execution
+- pattern_mining_category
+- pattern_mining_element
+- pattern_mining_config
+- pattern_itemset
+- pattern_itemset_item
+- pattern_itemset_post
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+from contextlib import contextmanager
+from datetime import date, datetime
+from typing import Any, Iterable, Mapping, Sequence
+
+try:
+    import psycopg2
+    from psycopg2.extras import RealDictCursor
+except Exception:  # pragma: no cover - exercised in environments without deps
+    psycopg2 = None
+    RealDictCursor = None
+
+
+PG_PATTERN_SOURCE_SYSTEM = "pg_pattern_v2"
+TOPIC_SCOPE = "topic"
+TOPIC_ELEMENT_SOURCE_TABLE = "post_decode_topic_point_element"
+
+_READONLY_SQL_RE = re.compile(r"^\s*(select|with|show|explain)\b", re.IGNORECASE | re.DOTALL)
+_BLOCKED_SQL_RE = re.compile(
+    r"\b(insert|update|delete|create|alter|drop|truncate|copy|grant|revoke|merge|call)\b",
+    re.IGNORECASE,
+)
+
+
+def _load_dotenv_once() -> None:
+    try:
+        from dotenv import load_dotenv
+    except Exception:
+        return
+    load_dotenv()
+
+
+def _resolve_dsn() -> str:
+    _load_dotenv_once()
+    dsn = os.getenv("DEMAND_PATTERN_PG_DSN") or os.getenv("PGVECTOR_DSN")
+    if not dsn:
+        raise RuntimeError("PG Pattern V2 DSN missing: set DEMAND_PATTERN_PG_DSN or PGVECTOR_DSN")
+    return dsn
+
+
+def _assert_readonly_sql(sql: str) -> None:
+    if not _READONLY_SQL_RE.search(sql):
+        raise RuntimeError("PG Pattern repository only allows SELECT/SHOW/EXPLAIN queries")
+    if _BLOCKED_SQL_RE.search(sql):
+        raise RuntimeError("PG Pattern repository blocked a non-read-only SQL keyword")
+
+
+@contextmanager
+def _connect_readonly():
+    if psycopg2 is None:
+        raise RuntimeError("psycopg2 is required for PG Pattern V2 reads")
+    conn = psycopg2.connect(_resolve_dsn())
+    try:
+        conn.set_session(readonly=True, autocommit=True)
+        yield conn
+    finally:
+        conn.close()
+
+
+def _fetch_all(sql: str, params: Sequence[Any] | Mapping[str, Any] | None = None) -> list[dict[str, Any]]:
+    _assert_readonly_sql(sql)
+    with _connect_readonly() as conn:
+        with conn.cursor(cursor_factory=RealDictCursor) as cursor:
+            cursor.execute(sql, params)
+            return [_row_to_dict(row) for row in cursor.fetchall()]
+
+
+def _fetch_one(sql: str, params: Sequence[Any] | Mapping[str, Any] | None = None) -> dict[str, Any] | None:
+    rows = _fetch_all(sql, params)
+    return rows[0] if rows else None
+
+
+def _cursor_fetch_all(cursor: Any, sql: str, params: Sequence[Any] | Mapping[str, Any] | None = None) -> list[dict[str, Any]]:
+    _assert_readonly_sql(sql)
+    cursor.execute(sql, params)
+    return [_row_to_dict(row) for row in cursor.fetchall()]
+
+
+def _cursor_fetch_one(cursor: Any, sql: str, params: Sequence[Any] | Mapping[str, Any] | None = None) -> dict[str, Any] | None:
+    rows = _cursor_fetch_all(cursor, sql, params)
+    return rows[0] if rows else None
+
+
+def _serialize_db_value(value: Any) -> Any:
+    if isinstance(value, (datetime, date)):
+        return value.isoformat()
+    return value
+
+
+def _row_to_dict(row: Any) -> dict[str, Any]:
+    if row is None:
+        return {}
+    mapping = getattr(row, "_mapping", row)
+    return {key: _serialize_db_value(value) for key, value in dict(mapping).items()}
+
+
+def _jsonish_to_list(value: Any) -> list[Any]:
+    if value is None:
+        return []
+    if isinstance(value, list):
+        return value
+    if isinstance(value, (tuple, set)):
+        return list(value)
+    if isinstance(value, bytes):
+        value = value.decode("utf-8")
+    if isinstance(value, str):
+        raw = value.strip()
+        if not raw:
+            return []
+        try:
+            parsed = json.loads(raw)
+        except json.JSONDecodeError:
+            return [part.strip() for part in raw.split(",") if part.strip()]
+        if isinstance(parsed, list):
+            return parsed
+        if parsed is None:
+            return []
+        return [parsed]
+    return [value]
+
+
+def _to_post_id_list(value: Any) -> list[str]:
+    post_ids: list[str] = []
+    seen: set[str] = set()
+    for item in _jsonish_to_list(value):
+        post_id = str(item).strip() if item is not None else ""
+        if post_id and post_id not in seen:
+            seen.add(post_id)
+            post_ids.append(post_id)
+    return post_ids
+
+
+def _to_int_list(values: Any) -> list[int]:
+    result: list[int] = []
+    seen: set[int] = set()
+    for value in _jsonish_to_list(values):
+        if value is None or value == "":
+            continue
+        try:
+            int_value = int(value)
+        except (TypeError, ValueError):
+            continue
+        if int_value not in seen:
+            seen.add(int_value)
+            result.append(int_value)
+    return result
+
+
+def _clean_names(names: Iterable[str]) -> list[str]:
+    result: list[str] = []
+    seen: set[str] = set()
+    for raw in names:
+        name = str(raw).strip() if raw is not None else ""
+        if name and name not in seen:
+            seen.add(name)
+            result.append(name)
+    return result
+
+
+def query_execution_for_evidence(execution_id: int) -> dict[str, Any] | None:
+    """Read PG Pattern V2 execution facts for evidence validation."""
+    row = _fetch_one(
+        """
+        SELECT
+            id,
+            snapshot_date,
+            is_current,
+            is_retired,
+            status,
+            post_filter,
+            post_count,
+            post_ids_digest,
+            classify_execution_id,
+            category_count,
+            element_count,
+            topic_itemset_count AS itemset_count,
+            topic_itemset_count,
+            topic_element_itemset_count,
+            cross_itemset_count,
+            script_sequence_count,
+            paragraph_itemset_count,
+            start_time,
+            end_time,
+            error_message
+        FROM pattern_mining_execution
+        WHERE id = %s
+        LIMIT 1
+        """,
+        (int(execution_id),),
+    )
+    if not row:
+        return None
+    row["pattern_source_system"] = PG_PATTERN_SOURCE_SYSTEM
+    return row
+
+
+def query_latest_success_execution_id() -> int | None:
+    row = _fetch_one(
+        """
+        SELECT id
+        FROM pattern_mining_execution
+        WHERE status = 'success'
+          AND COALESCE(topic_itemset_count, 0) > 0
+        ORDER BY snapshot_date DESC NULLS LAST, id DESC
+        LIMIT 1
+        """
+    )
+    return int(row["id"]) if row else None
+
+
+def query_itemset_evidence(execution_id: int, itemset_ids: Iterable[Any]) -> list[dict[str, Any]]:
+    """Read PG topic-scope itemset facts and aggregate matched posts."""
+    clean_ids = _to_int_list(itemset_ids or [])
+    if not clean_ids:
+        return []
+
+    rows = _fetch_all(
+        """
+        SELECT
+            i.id,
+            i.execution_id,
+            i.mining_config_id,
+            cfg.execution_id AS mining_config_execution_id,
+            cfg.scope AS mining_config_scope,
+            cfg.name AS mining_config_name,
+            cfg.algorithm AS mining_config_algorithm,
+            cfg.params::text AS mining_config_params,
+            cfg.params ->> 'dimension_mode' AS mining_config_dimension_mode,
+            COALESCE(cfg.params ->> 'target_depth', cfg.params ->> 'depth') AS mining_config_target_depth,
+            i.scope,
+            i.combination_type,
+            i.item_count,
+            i.support,
+            i.absolute_support,
+            i.dimensions::text AS dimensions,
+            i.is_cross_point,
+            i.is_cross_layer,
+            COALESCE(
+                array_agg(p.post_id ORDER BY p.post_id)
+                    FILTER (WHERE p.post_id IS NOT NULL),
+                ARRAY[]::varchar[]
+            ) AS matched_post_ids
+        FROM pattern_itemset i
+        JOIN pattern_mining_config cfg
+          ON cfg.id = i.mining_config_id
+         AND cfg.execution_id = i.execution_id
+         AND cfg.scope = %s
+        LEFT JOIN pattern_itemset_post p
+          ON p.itemset_id = i.id
+         AND p.execution_id = i.execution_id
+        WHERE i.execution_id = %s
+          AND i.scope = %s
+          AND i.id = ANY(%s)
+        GROUP BY
+            i.id,
+            i.execution_id,
+            i.mining_config_id,
+            cfg.execution_id,
+            cfg.scope,
+            cfg.name,
+            cfg.algorithm,
+            cfg.params::text,
+            cfg.params ->> 'dimension_mode',
+            COALESCE(cfg.params ->> 'target_depth', cfg.params ->> 'depth'),
+            i.scope,
+            i.combination_type,
+            i.item_count,
+            i.support,
+            i.absolute_support,
+            i.dimensions::text,
+            i.is_cross_point,
+            i.is_cross_layer
+        ORDER BY i.id
+        """,
+        (TOPIC_SCOPE, int(execution_id), TOPIC_SCOPE, clean_ids),
+    )
+
+    result: list[dict[str, Any]] = []
+    for row in rows:
+        itemset = dict(row)
+        for key in (
+                "id",
+                "execution_id",
+                "mining_config_id",
+                "mining_config_execution_id",
+                "item_count",
+                "absolute_support",
+        ):
+            if itemset.get(key) is not None:
+                itemset[key] = int(itemset[key])
+        itemset["support"] = float(itemset["support"]) if itemset.get("support") is not None else None
+        itemset["is_cross_point"] = bool(itemset.get("is_cross_point"))
+        itemset["is_cross_layer"] = bool(itemset.get("is_cross_layer"))
+        itemset["dimensions"] = _jsonish_to_list(itemset.get("dimensions"))
+        if isinstance(itemset.get("mining_config_params"), str):
+            try:
+                itemset["mining_config_params"] = json.loads(itemset["mining_config_params"])
+            except json.JSONDecodeError:
+                pass
+        itemset["matched_post_ids"] = _to_post_id_list(itemset.get("matched_post_ids"))
+        result.append(itemset)
+
+    by_id = {itemset["id"]: itemset for itemset in result}
+    return [by_id[itemset_id] for itemset_id in clean_ids if itemset_id in by_id]
+
+
+def query_itemset_items_with_categories(
+        execution_id: int,
+        itemset_ids: Iterable[Any],
+) -> list[dict[str, Any]]:
+    """Read PG itemset items and their exact category tree nodes."""
+    clean_ids = _to_int_list(itemset_ids or [])
+    if not clean_ids:
+        return []
+
+    rows = _fetch_all(
+        """
+        SELECT
+            ii.id AS itemset_item_id,
+            ii.itemset_id,
+            ii.layer,
+            ii.point_type,
+            ii.dimension,
+            ii.category_id,
+            ii.category_path,
+            ii.element_name,
+            ii.element_id,
+            ii.post_count,
+            c.id AS bound_category_id,
+            c.execution_id AS category_execution_id,
+            c.source_stable_id AS category_source_stable_id,
+            c.source_type AS category_source_type,
+            c.name AS category_name,
+            c.path AS category_full_path,
+            c.level AS category_level,
+            c.parent_id AS category_parent_id,
+            c.parent_source_stable_id AS category_parent_source_stable_id,
+            c.classified_as AS category_nature
+        FROM pattern_itemset_item ii
+        JOIN pattern_itemset i
+          ON i.id = ii.itemset_id
+         AND i.execution_id = %s
+         AND i.scope = %s
+        LEFT JOIN pattern_mining_category c
+          ON c.id = ii.category_id
+         AND c.execution_id = i.execution_id
+        WHERE ii.itemset_id = ANY(%s)
+        ORDER BY ii.itemset_id, ii.id
+        """,
+        (int(execution_id), TOPIC_SCOPE, clean_ids),
+    )
+    result: list[dict[str, Any]] = []
+    for row in rows:
+        item = dict(row)
+        for key in (
+                "itemset_item_id",
+                "itemset_id",
+                "category_id",
+                "element_id",
+                "post_count",
+                "bound_category_id",
+                "category_execution_id",
+                "category_source_stable_id",
+                "category_level",
+                "category_parent_id",
+                "category_parent_source_stable_id",
+        ):
+            if item.get(key) is not None:
+                item[key] = int(item[key])
+        item["category_found"] = item.get("bound_category_id") is not None
+        result.append(item)
+    return result
+
+
+def query_element_bindings_for_items(
+        execution_id: int,
+        itemset_items: Sequence[Mapping[str, Any]],
+        post_ids: Iterable[Any] | None = None,
+        limit_per_item: int = 200,
+) -> list[dict[str, Any]]:
+    """Bind itemset items to real PG Pattern V2 source-post elements."""
+    clean_post_ids = _to_post_id_list(post_ids)
+    if not itemset_items:
+        return []
+
+    bindings: list[dict[str, Any]] = []
+    with _connect_readonly() as conn:
+        with conn.cursor(cursor_factory=RealDictCursor) as cursor:
+            for raw_item in itemset_items:
+                item = dict(raw_item)
+                category_id = item.get("bound_category_id") or item.get("category_id")
+                if category_id is None:
+                    continue
+
+                clauses = [
+                    "execution_id = %s",
+                    "category_id = %s",
+                    "source_table = %s",
+                ]
+                params: list[Any] = [int(execution_id), int(category_id), TOPIC_ELEMENT_SOURCE_TABLE]
+
+                if item.get("point_type"):
+                    clauses.append("point_type = %s")
+                    params.append(item["point_type"])
+                if item.get("dimension"):
+                    clauses.append("element_type = %s")
+                    params.append(item["dimension"])
+                if item.get("element_name"):
+                    clauses.append("name = %s")
+                    params.append(item["element_name"])
+                if clean_post_ids:
+                    clauses.append("post_id = ANY(%s)")
+                    params.append(clean_post_ids)
+
+                where_sql = " AND ".join(clauses)
+                count_row = _cursor_fetch_one(
+                    cursor,
+                    f"""
+                    SELECT
+                        COUNT(*) AS matched_element_count,
+                        COUNT(DISTINCT post_id) AS matched_post_count
+                    FROM pattern_mining_element
+                    WHERE {where_sql}
+                    """,
+                    tuple(params),
+                ) or {}
+                post_rows = _cursor_fetch_all(
+                    cursor,
+                    f"""
+                    SELECT DISTINCT post_id
+                    FROM pattern_mining_element
+                    WHERE {where_sql}
+                    ORDER BY post_id
+                    LIMIT %s
+                    """,
+                    tuple(params + [int(limit_per_item)]),
+                )
+                sample_rows = _cursor_fetch_all(
+                    cursor,
+                    f"""
+                    SELECT
+                        id,
+                        post_id,
+                        source_table,
+                        source_element_id,
+                        point_type,
+                        point_text,
+                        element_type,
+                        name,
+                        category_id,
+                        category_path,
+                        topic_point_id
+                    FROM pattern_mining_element
+                    WHERE {where_sql}
+                    ORDER BY post_id, id
+                    LIMIT %s
+                    """,
+                    tuple(params + [min(int(limit_per_item), 20)]),
+                )
+
+                bindings.append(
+                    {
+                        "itemset_id": item.get("itemset_id"),
+                        "itemset_item_id": item.get("itemset_item_id"),
+                        "category_id": int(category_id),
+                        "point_type": item.get("point_type"),
+                        "dimension": item.get("dimension"),
+                        "element_name": item.get("element_name"),
+                        "matched_element_count": int(count_row.get("matched_element_count") or 0),
+                        "matched_post_count": int(count_row.get("matched_post_count") or 0),
+                        "matched_post_ids": _to_post_id_list([row["post_id"] for row in post_rows]),
+                        "sample_elements": sample_rows,
+                    }
+                )
+    return bindings
+
+
+def query_video_ids_by_names(execution_id: int, names: Iterable[str]) -> list[str]:
+    """Resolve names to PG source post IDs by category name/path leaf or element name."""
+    clean_names = _clean_names(names)
+    if not clean_names:
+        return []
+    rows = _fetch_all(
+        """
+        WITH target_categories AS (
+            SELECT id
+            FROM pattern_mining_category
+            WHERE execution_id = %s
+              AND (
+                name = ANY(%s)
+                OR regexp_replace(path, '^.*/', '') = ANY(%s)
+              )
+        ),
+        matched_elements AS (
+            SELECT post_id
+            FROM pattern_mining_element
+            WHERE execution_id = %s
+              AND source_table = %s
+              AND (
+                name = ANY(%s)
+                OR category_id IN (SELECT id FROM target_categories)
+              )
+        )
+        SELECT DISTINCT post_id
+        FROM matched_elements
+        WHERE post_id IS NOT NULL
+        ORDER BY post_id
+        """,
+        (int(execution_id), clean_names, clean_names, int(execution_id), TOPIC_ELEMENT_SOURCE_TABLE, clean_names),
+    )
+    return _to_post_id_list([row["post_id"] for row in rows])
+
+
+def query_category_level(execution_id: int, name: str) -> int | None:
+    clean_name = str(name).strip() if name is not None else ""
+    if not clean_name:
+        return None
+    row = _fetch_one(
+        """
+        SELECT level
+        FROM pattern_mining_category
+        WHERE execution_id = %s
+          AND (name = %s OR regexp_replace(path, '^.*/', '') = %s)
+        ORDER BY id DESC
+        LIMIT 1
+        """,
+        (int(execution_id), clean_name, clean_name),
+    )
+    return int(row["level"]) if row and row.get("level") is not None else None
+
+
+def query_case_ids_by_post_ids(post_ids: Iterable[Any]) -> list[dict[str, Any]]:
+    """PG Pattern V2 has post_id semantics for MVP exact evidence.
+
+    DemandAgent no longer reads the old MySQL decode-case table in this PG-only
+    path. `case_ids` is filled from validated post IDs by the evidence builder;
+    `decode_case_ids` remains an optional empty compatibility field.
+    """
+    return []
+
+
+def query_categories(
+        execution_id: int,
+        *,
+        source_type: str | None = None,
+        keyword: str | None = None,
+        category_ids: Iterable[Any] | None = None,
+        limit: int | None = None,
+) -> list[dict[str, Any]]:
+    clauses = ["execution_id = %s"]
+    params: list[Any] = [int(execution_id)]
+    if source_type:
+        clauses.append("source_type = %s")
+        params.append(source_type)
+    if keyword:
+        clauses.append("(name ILIKE %s OR path ILIKE %s)")
+        like = f"%{keyword}%"
+        params.extend([like, like])
+    clean_ids = _to_int_list(category_ids or [])
+    if clean_ids:
+        clauses.append("id = ANY(%s)")
+        params.append(clean_ids)
+
+    limit_sql = "LIMIT %s" if limit else ""
+    if limit:
+        params.append(int(limit))
+    return _fetch_all(
+        f"""
+        SELECT
+            id,
+            execution_id,
+            source_stable_id,
+            source_type,
+            name,
+            description,
+            path,
+            level,
+            parent_id,
+            parent_source_stable_id,
+            element_count,
+            classified_as AS category_nature
+        FROM pattern_mining_category
+        WHERE {" AND ".join(clauses)}
+        ORDER BY source_type, path, id
+        {limit_sql}
+        """,
+        tuple(params),
+    )
+
+
+def query_elements(
+        execution_id: int,
+        *,
+        keyword: str | None = None,
+        element_type: str | None = None,
+        post_ids: Iterable[Any] | None = None,
+        category_ids: Iterable[Any] | None = None,
+        limit: int | None = None,
+) -> list[dict[str, Any]]:
+    clauses = [
+        "execution_id = %s",
+        "source_table = %s",
+    ]
+    params: list[Any] = [int(execution_id), TOPIC_ELEMENT_SOURCE_TABLE]
+    if keyword:
+        clauses.append("name ILIKE %s")
+        params.append(f"%{keyword}%")
+    if element_type:
+        clauses.append("element_type = %s")
+        params.append(element_type)
+    clean_posts = _to_post_id_list(post_ids)
+    if clean_posts:
+        clauses.append("post_id = ANY(%s)")
+        params.append(clean_posts)
+    clean_categories = _to_int_list(category_ids or [])
+    if clean_categories:
+        clauses.append("category_id = ANY(%s)")
+        params.append(clean_categories)
+
+    limit_sql = "LIMIT %s" if limit else ""
+    if limit:
+        params.append(int(limit))
+    return _fetch_all(
+        f"""
+        SELECT
+            id,
+            execution_id,
+            post_id,
+            source_table,
+            source_element_id,
+            element_type,
+            element_sub_type,
+            name,
+            description,
+            category_id,
+            category_path,
+            point_type,
+            point_text,
+            topic_point_id
+        FROM pattern_mining_element
+        WHERE {" AND ".join(clauses)}
+        ORDER BY post_id, id
+        {limit_sql}
+        """,
+        tuple(params),
+    )
+
+
+def query_topic_itemsets(
+        execution_id: int,
+        *,
+        category_ids: Iterable[Any] | None = None,
+        dimension_mode: str | None = None,
+        min_support: int | None = None,
+        min_item_count: int | None = None,
+        max_item_count: int | None = None,
+        sort_by: str = "absolute_support",
+        limit: int = 100,
+) -> list[dict[str, Any]]:
+    clauses = [
+        "i.execution_id = %s",
+        "i.scope = %s",
+    ]
+    params: list[Any] = [int(execution_id), TOPIC_SCOPE]
+    clean_category_ids = _to_int_list(category_ids or [])
+    for category_id in clean_category_ids:
+        clauses.append(
+            """
+            EXISTS (
+                SELECT 1
+                FROM pattern_itemset_item ii_filter
+                WHERE ii_filter.itemset_id = i.id
+                  AND ii_filter.category_id = %s
+            )
+            """
+        )
+        params.append(category_id)
+    if dimension_mode:
+        clauses.append("cfg.params ->> 'dimension_mode' = %s")
+        params.append(dimension_mode)
+    if min_support is not None:
+        clauses.append("i.absolute_support >= %s")
+        params.append(int(min_support))
+    if min_item_count is not None:
+        clauses.append("i.item_count >= %s")
+        params.append(int(min_item_count))
+    if max_item_count is not None:
+        clauses.append("i.item_count <= %s")
+        params.append(int(max_item_count))
+
+    order_sql = {
+        "support": "i.support DESC NULLS LAST, i.absolute_support DESC NULLS LAST",
+        "item_count": "i.item_count DESC NULLS LAST, i.absolute_support DESC NULLS LAST",
+        "absolute_support": "i.absolute_support DESC NULLS LAST, i.support DESC NULLS LAST",
+    }.get(str(sort_by or "absolute_support"), "i.absolute_support DESC NULLS LAST, i.support DESC NULLS LAST")
+    params.append(int(limit))
+    return _fetch_all(
+        f"""
+        SELECT
+            i.id,
+            i.execution_id,
+            i.mining_config_id,
+            cfg.scope AS mining_config_scope,
+            cfg.name AS mining_config_name,
+            cfg.params AS mining_config_params,
+            cfg.params ->> 'dimension_mode' AS dimension_mode,
+            COALESCE(cfg.params ->> 'target_depth', cfg.params ->> 'depth') AS target_depth,
+            i.scope,
+            i.combination_type,
+            i.item_count,
+            i.support,
+            i.absolute_support,
+            i.dimensions,
+            i.is_cross_point,
+            i.is_cross_layer
+        FROM pattern_itemset i
+        JOIN pattern_mining_config cfg
+          ON cfg.id = i.mining_config_id
+         AND cfg.execution_id = i.execution_id
+         AND cfg.scope = %s
+        WHERE {" AND ".join(clauses)}
+        ORDER BY {order_sql}, i.id
+        LIMIT %s
+        """,
+        tuple([TOPIC_SCOPE, *params]),
+    )
+
+
+def query_weight_score_rows(
+        execution_id: int,
+        *,
+        level: str,
+        dimension: str,
+        limit: int = 5000,
+) -> list[dict[str, Any]]:
+    """Aggregate PG topic elements into old weight-score JSON shapes."""
+    if level == "元素":
+        return _fetch_all(
+            """
+            SELECT
+                name,
+                element_type,
+                category_id,
+                category_path,
+                COUNT(*) AS occurrence_count,
+                COUNT(DISTINCT post_id) AS post_count
+            FROM pattern_mining_element
+            WHERE execution_id = %s
+              AND source_table = %s
+              AND element_type = %s
+              AND name IS NOT NULL
+              AND name <> ''
+            GROUP BY name, element_type, category_id, category_path
+            ORDER BY COUNT(DISTINCT post_id) DESC, COUNT(*) DESC, name
+            LIMIT %s
+            """,
+            (int(execution_id), TOPIC_ELEMENT_SOURCE_TABLE, dimension, int(limit)),
+        )
+    if level == "分类":
+        return _fetch_all(
+            """
+            SELECT
+                COALESCE(c.name, regexp_replace(e.category_path, '^.*/', '')) AS category,
+                e.category_id,
+                e.category_path,
+                e.element_type,
+                COUNT(*) AS occurrence_count,
+                COUNT(DISTINCT e.post_id) AS post_count,
+                COALESCE(MAX(c.level), array_length(string_to_array(trim(both '/' from e.category_path), '/'), 1)) AS level
+            FROM pattern_mining_element e
+            LEFT JOIN pattern_mining_category c
+              ON c.id = e.category_id
+             AND c.execution_id = e.execution_id
+            WHERE e.execution_id = %s
+              AND e.source_table = %s
+              AND e.element_type = %s
+              AND e.category_id IS NOT NULL
+              AND e.category_path IS NOT NULL
+            GROUP BY e.category_id, e.category_path, e.element_type, c.name
+            ORDER BY COUNT(DISTINCT e.post_id) DESC, COUNT(*) DESC, e.category_path
+            LIMIT %s
+            """,
+            (int(execution_id), TOPIC_ELEMENT_SOURCE_TABLE, dimension, int(limit)),
+        )
+    raise ValueError(f"unsupported weight score level: {level}")

examples/demand/pg_weight_score_builder.py

@@ -0,0 +1,82 @@
+"""Build local weight-score JSON files from PG Pattern V2 facts."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from examples.demand.pg_pattern_repository import PG_PATTERN_SOURCE_SYSTEM, query_weight_score_rows
+
+
+DIMENSIONS = ("实质", "形式", "意图")
+LEVELS = ("元素", "分类")
+
+
+def _score(post_count: Any, occurrence_count: Any) -> float:
+    """Stable MVP score from real PG coverage counts."""
+    return float(post_count or 0) + float(occurrence_count or 0) / 1000000.0
+
+
+def _normalize_element_row(row: dict[str, Any]) -> dict[str, Any]:
+    occurrence_count = int(row.get("occurrence_count") or 0)
+    post_count = int(row.get("post_count") or 0)
+    return {
+        "name": row.get("name"),
+        "element_type": row.get("element_type"),
+        "category_id": row.get("category_id"),
+        "category_path": row.get("category_path"),
+        "occurrence_count": occurrence_count,
+        "post_count": post_count,
+        "score": _score(post_count, occurrence_count),
+        "source_system": PG_PATTERN_SOURCE_SYSTEM,
+    }
+
+
+def _normalize_category_row(row: dict[str, Any]) -> dict[str, Any]:
+    occurrence_count = int(row.get("occurrence_count") or 0)
+    post_count = int(row.get("post_count") or 0)
+    return {
+        "category": row.get("category"),
+        "category_id": row.get("category_id"),
+        "category_path": row.get("category_path"),
+        "element_type": row.get("element_type"),
+        "level": row.get("level"),
+        "occurrence_count": occurrence_count,
+        "post_count": post_count,
+        "score": _score(post_count, occurrence_count),
+        "source_system": PG_PATTERN_SOURCE_SYSTEM,
+    }
+
+
+def build_pg_weight_score_files(
+        execution_id: int,
+        *,
+        base_dir: str | Path,
+        limit_per_file: int = 5000,
+) -> dict[str, int]:
+    """Write old-shape weight JSON files under `{base_dir}/{execution_id}`."""
+    output_dir = Path(base_dir) / str(execution_id)
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+    counts: dict[str, int] = {}
+    for dimension in DIMENSIONS:
+        for level in LEVELS:
+            rows = query_weight_score_rows(
+                execution_id=execution_id,
+                level=level,
+                dimension=dimension,
+                limit=limit_per_file,
+            )
+            if level == "元素":
+                payload = [_normalize_element_row(dict(row)) for row in rows]
+            else:
+                payload = [_normalize_category_row(dict(row)) for row in rows]
+            payload.sort(key=lambda item: float(item.get("score") or 0), reverse=True)
+            path = output_dir / f"{dimension}_{level}.json"
+            path.write_text(
+                json.dumps(payload, ensure_ascii=False, indent=2, default=str) + "\n",
+                encoding="utf-8",
+            )
+            counts[f"{dimension}_{level}"] = len(payload)
+    return counts

examples/demand/run.py

@@ -1,7 +1,6 @@
 """demand 示例的最小可运行入口。"""
 import asyncio
 import copy
-import hashlib
 import importlib
 import json
 import os
@@ -22,10 +21,11 @@ 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.db_manager import (
+    query_category_level,
+    query_latest_success_execution_id,
+    query_video_ids_by_names,
+)
 from examples.demand.demand_agent_context import TopicBuildAgentContext
 
 # 添加项目根目录到 Python 路径
@@ -65,6 +65,13 @@ def _is_local_json_mode() -> bool:
     return os.getenv("DEMAND_OUTPUT_MODE", "").strip().lower() == LOCAL_JSON_MODE
 
 
+def _raise_db_write_disabled(action: str) -> None:
+    raise RuntimeError(
+        f"{action} 已在 PG Pattern V2 local MVP 中禁用；"
+        "请通过 DEMAND_OUTPUT_MODE=local_json 和 run_existing_execution_local 写本地 JSON"
+    )
+
+
 def _require_local_entrypoint() -> None:
     if not _is_local_json_mode():
         return
@@ -86,6 +93,7 @@ def _ensure_local_output_paths(execution_id: int) -> Path:
     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
@@ -120,6 +128,13 @@ def _get_trace_store_base_dir() -> 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
@@ -132,23 +147,22 @@ def _write_json_file(path: Path, payload: Any) -> None:
 
 
 def get_execution_id_by_merge_level2(cluster_name: str):
-    """根据二级品类和平台查询最新的 execution_id。"""
-    session = DatabaseManager().get_session()
-    try:
-        start_of_today = datetime.combine(datetime.now().date(), datetime.min.time())
-        query = session.query(TopicPatternExecution).filter(
-            TopicPatternExecution.cluster_name == cluster_name,
-            TopicPatternExecution.start_time >= start_of_today,
-        )
-        execution = query.order_by(desc(TopicPatternExecution.id)).first()
-        if not execution:
-            return None
-        return execution.id
-    finally:
-        session.close()
+    """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
@@ -213,7 +227,7 @@ def _load_name_score_map(execution_id: int) -> dict:
     - `*_元素.json`：字段 `name` 表示名字
     - `*_分类.json`：字段 `category` 表示名字
     """
-    data_dir = Path(__file__).parent / "data" / str(execution_id)
+    data_dir = _get_weight_data_base_dir() / str(execution_id)
     if not data_dir.exists():
         return {}
 
@@ -244,6 +258,25 @@ def _load_name_score_map(execution_id: int) -> dict:
     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():
+        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()]
@@ -313,18 +346,6 @@ def _normalize_evidence_refs(value: object) -> dict:
     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,
@@ -433,110 +454,22 @@ def _build_demand_content_rows(
     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]
-        # 数据库字段 demand_task.platform: varchar(32)
-        if platform is not None:
-            platform = str(platform)[:32]
-
-        task_id = mysql_db.insert(
-            "demand_task",
-            {
-                "execution_id": execution_id,
-                "name": name,
-                "platform": platform,
-                "status": 0,
-                "log": "",
-            },
-        )
-        log(f"[task] 创建 demand_task 成功，task_id={task_id}, execution_id={execution_id}")
-        return task_id
-    except Exception as e:
-        log(f"[task] 创建 demand_task 失败，execution_id={execution_id}, error={e}")
-        return None
+    """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:
-    """更新 demand_task 状态与日志。"""
+    """Disabled production demand_task updater retained for legacy imports."""
     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",
-            {
-                "status": int(status),
-                "log": task_log or "",
-            },
-            "id = %s",
-            (task_id,),
-        )
-        log(f"[task] 更新 demand_task 成功，task_id={task_id}, status={status}")
-    except Exception as e:
-        log(f"[task] 更新 demand_task 失败，task_id={task_id}, status={status}, error={e}")
+    del status, task_log
+    _raise_db_write_disabled("生产 _finish_demand_task")
 
 
 def _start_local_demand_task(
@@ -616,41 +549,9 @@ def write_demand_items_to_mysql(
         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")
-
-    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:
-        return 0
-
-    affected = mysql_db.insert_many("demand_content", rows)
-    log(f"[mysql] 写入 demand_content 完成，rows={len(rows)}, affected={affected}")
-
-    try:
-        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}")
-    # with open(f'/Users/shimeng/Desktop/py/Agent/examples/demand/result/{merge_level2}.json', 'w',
-    #           encoding='utf-8') as f:
-    #     json.dump(rows, f, ensure_ascii=False, indent=4)
-    return len(rows)
+    """Disabled production demand_content writer retained for legacy imports."""
+    del execution_id, merge_level2, trace_id, task_id
+    _raise_db_write_disabled("生产 write_demand_items_to_mysql")
 
 
 def write_demand_items_to_local_json(
@@ -687,39 +588,9 @@ def write_demand_items_to_local_json(
 
 
 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（仅全局树场景）。
-    """
-    if _is_local_json_mode():
-        raise RuntimeError("local_json 模式禁止调用生产 write_global_tree_demand_items_to_hive")
-
-    items = _load_demand_items(execution_id=execution_id, log_prefix="hive")
-    if not items:
-        return 0
-
-    names: list[str] = []
-    for di in items:
-        if not isinstance(di, dict):
-            continue
-        name = _join_element_names_to_name(di.get("element_names"))
-        if name:
-            names.append(name)
-
-    if not names:
-        log("[hive] 无有效 name，跳过写入")
-        return 0
-
-    try:
-        # 按用户要求：Hive 写入逻辑放在 data_query_tools.py 中。
-        from examples.demand.data_query_tools import write_feature_point_data_to_hive
-
-        written = write_feature_point_data_to_hive(names=names)
-        log(f"[hive] 写入 feature_point_data 完成，rows={written}")
-        return written
-    except Exception as e:
-        log(f"[hive] 写入 feature_point_data 异常：{e}")
-        return 0
+    """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(
@@ -824,6 +695,7 @@ async def run_once(
 
     if _is_local_json_mode():
         _ensure_local_output_paths(execution_id)
+        _ensure_pg_weight_score_files(int(execution_id))
 
     TopicBuildAgentContext.set_execution_id(execution_id)
     TopicBuildAgentContext.set_metadata("result_base_dir", str(_get_result_base_dir()))
@@ -995,6 +867,9 @@ async def main(
         execution_id: Optional[int] = None,
         task_id: Optional[int] = None,
 ) -> dict:
+    if not _is_local_json_mode():
+        _raise_db_write_disabled("非 local_json 运行入口")
+
     if _is_local_json_mode():
         _require_local_entrypoint()
         if execution_id is None:
@@ -1002,20 +877,7 @@ async def main(
         _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
+        execution_id = get_execution_id_by_merge_level2(cluster_name)
     if not execution_id:
         return {"execution_id": None, "trace_id": None, "final_text": ""}
 

examples/demand/run_existing_execution_local.py

@@ -1,7 +1,8 @@
-"""Local JSON entrypoint for an existing DemandAgent execution.
+"""Local JSON entrypoint for an existing PG Pattern V2 execution.
 
-This entrypoint intentionally requires an existing execution_id. It redirects
-result/.trace/output into one local run directory and never calls prepare.
+This entrypoint intentionally requires an existing PG pattern_mining_execution.id.
+It redirects result/.trace/output into one local run directory and never calls
+prepare/run_mining.
 """
 import argparse
 import asyncio
@@ -45,6 +46,7 @@ def _configure_local_env(execution_id: int, output_root: Path) -> None:
     os.environ["DEMAND_LOCAL_ENTRYPOINT"] = "run_existing_execution_local"
     os.environ["DEMAND_LOCAL_OUTPUT_ROOT"] = str(output_root)
     os.environ["DEMAND_RESULT_BASE_DIR"] = str(output_root / "intermediate" / "result")
+    os.environ["DEMAND_WEIGHT_DATA_DIR"] = str(output_root / "intermediate" / "data")
     os.environ["DEMAND_TRACE_STORE_PATH"] = str(output_root / ".trace")
     os.environ["DEMAND_OUTPUT_BASE_DIR"] = str(output_root / "output")
     os.environ.setdefault("DEMAND_LOCAL_TASK_ID", "1")
@@ -64,9 +66,9 @@ def _validate_execution_success(execution_id: int) -> None:
 
 def parse_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser(
-        description="Run DemandAgent against an existing execution_id and write local JSON outputs."
+        description="Run DemandAgent against an existing PG Pattern V2 execution_id and write local JSON outputs."
     )
-    parser.add_argument("--execution-id", type=int, required=True, help="Existing topic_pattern_execution.id")
+    parser.add_argument("--execution-id", type=int, required=True, help="Existing PG pattern_mining_execution.id")
     parser.add_argument("--merge-level2", required=True, help="Current merge_leve2 / cluster name")
     parser.add_argument("--platform-type", default="piaoquan", help="Platform label for manifest only")
     parser.add_argument("--count", type=int, default=5, help="Approximate demand count requested from the prompt")

examples/demand/tests/__init__.py

@@ -0,0 +1 @@
+

examples/demand/tests/test_pg_evidence_builder.py

@@ -0,0 +1,131 @@
+import unittest
+from unittest.mock import patch
+
+from examples.demand.evidence_pack_builder import build_evidence_pack
+
+
+def _execution():
+    return {"id": 581, "status": "success"}
+
+
+def _itemset(**overrides):
+    data = {
+        "id": 1607313,
+        "execution_id": 581,
+        "mining_config_id": 2081,
+        "mining_config_execution_id": 581,
+        "mining_config_scope": "topic",
+        "scope": "topic",
+        "item_count": 1,
+        "support": 0.1,
+        "absolute_support": 2,
+        "matched_post_ids": ["p1", "p2"],
+    }
+    data.update(overrides)
+    return data
+
+
+def _item():
+    return {
+        "itemset_item_id": 1,
+        "itemset_id": 1607313,
+        "category_id": 10,
+        "bound_category_id": 10,
+        "category_found": True,
+        "category_execution_id": 581,
+        "category_name": "综合性腐败",
+        "category_path": "/事件行为/违纪违法/综合性腐败",
+        "category_full_path": "/事件行为/违纪违法/综合性腐败",
+        "dimension": "实质",
+        "point_type": "关键点",
+        "element_name": None,
+    }
+
+
+def _binding():
+    return {
+        "itemset_id": 1607313,
+        "itemset_item_id": 1,
+        "category_id": 10,
+        "dimension": "实质",
+        "point_type": "关键点",
+        "element_name": None,
+        "matched_element_count": 1,
+        "matched_post_count": 1,
+        "matched_post_ids": ["p1"],
+        "sample_elements": [{"name": "综合性腐败", "category_path": "/事件行为/违纪违法/综合性腐败"}],
+    }
+
+
+class PgEvidenceBuilderTest(unittest.TestCase):
+    def _build(self, evidence_refs):
+        return build_evidence_pack(
+            581,
+            {"element_names": ["综合性腐败"], "evidence_refs": evidence_refs},
+            trace_id="trace-test",
+            demand_task_id=1,
+            demand_content_id=1,
+        )
+
+    def _patch_success(self, itemsets=None):
+        patches = [
+            patch("examples.demand.evidence_pack_builder.query_execution_for_evidence", return_value=_execution()),
+            patch("examples.demand.evidence_pack_builder.query_itemset_evidence", return_value=itemsets or [_itemset()]),
+            patch("examples.demand.evidence_pack_builder.query_itemset_items_with_categories", return_value=[_item()]),
+            patch("examples.demand.evidence_pack_builder.query_element_bindings_for_items", return_value=[_binding()]),
+            patch("examples.demand.evidence_pack_builder.query_case_ids_by_post_ids", return_value=[]),
+        ]
+        for patcher in patches:
+            patcher.start()
+            self.addCleanup(patcher.stop)
+
+    def test_topic_itemset_builds_pg_evidence_pack(self):
+        self._patch_success()
+        result = self._build(
+            {
+                "source_kind": "pattern_itemset",
+                "source_tool": "get_itemset_detail",
+                "itemset_ids": [1607313],
+                "source_post_id": "p1",
+                "seed_terms": ["综合性腐败"],
+            }
+        )
+
+        self.assertTrue(result["success"])
+        evidence_pack = result["evidence_pack"]
+        self.assertEqual(evidence_pack["pattern_source_system"], "pg_pattern_v2")
+        self.assertEqual(evidence_pack["validation_status"], "passed")
+        self.assertEqual(evidence_pack["source_certainty"], "db_validated")
+        self.assertEqual(evidence_pack["matched_post_ids"], ["p1", "p2"])
+
+    def test_non_topic_scope_is_rejected(self):
+        self._patch_success(itemsets=[_itemset(scope="topic_element")])
+        result = self._build(
+            {
+                "source_kind": "pattern_itemset",
+                "itemset_ids": [1607313],
+                "source_post_id": "p1",
+                "seed_terms": ["综合性腐败"],
+            }
+        )
+
+        self.assertFalse(result["success"])
+        self.assertIn("scope=topic_element", result["reject_reason"])
+
+    def test_source_post_outside_matched_posts_is_rejected(self):
+        self._patch_success()
+        result = self._build(
+            {
+                "source_kind": "pattern_itemset",
+                "itemset_ids": [1607313],
+                "source_post_id": "not-a-source-post",
+                "seed_terms": ["综合性腐败"],
+            }
+        )
+
+        self.assertFalse(result["success"])
+        self.assertIn("not in matched_post_ids", result["reject_reason"])
+
+
+if __name__ == "__main__":
+    unittest.main()

examples/demand/weight_score_query_tools.py

@@ -7,6 +7,7 @@
 支持按元素/分类查询 TopN，以及按名称列表批量查询权重分。
 """
 import json
+import os
 from pathlib import Path
 
 from agent import tool
@@ -24,7 +25,11 @@ def _get_weight_file_path(level: str, dimension: str) -> Path:
         raise ValueError("未设置 execution_id，请先在 TopicBuildAgentContext 中设置")
 
     filename = f"{dimension}_{level}.json"
-    base_dir = Path(__file__).parent / "data" / str(execution_id)
+    configured_base_dir = os.getenv("DEMAND_WEIGHT_DATA_DIR")
+    if configured_base_dir:
+        base_dir = Path(configured_base_dir) / str(execution_id)
+    else:
+        base_dir = Path(__file__).parent / "data" / str(execution_id)
     return base_dir / filename
 
 

requirements.txt

@@ -22,6 +22,7 @@ fastapi==0.135.2
 pydantic==2.12.5
 sqlalchemy==2.0.48
 pymysql==1.1.2
+psycopg2-binary>=2.9.9
 requests==2.33.0
 pandas==3.0.1
 pyfim==6.28