接入成员画像 Dify 工作流并清理旧提取逻辑

- 新增 member_context 专用 DifyClient，统一兼容 completion 与 workflow 两种调用模式
- 将成员画像插件默认切换到 Dify workflow 模式，配置改用新的 workflow 应用与 workflows/run 接口
- 生成可直接导入 Dify 的成员画像工作流 DSL 文件，方便后台一键导入和发布
- 补充 Dify 工作流接入说明文档，明确输入字段、输出字段、发布步骤与插件消费方式
- 清理旧的单成员日摘要提取链路，日级画像统一收敛到群日批量提取路径，减少无效分支和历史残留
- 去除 member_context 内部多处旧 requests 直连调用，统一改为通过 DifyClient 调用 AI 服务
- 优化群日批量结果解析逻辑，只按 wxid 作为唯一主键识别成员，不再依赖昵称做唯一判断
- 新增按 wxid 的结果去重与完整度评分逻辑，遇到重复成员结果时优先保留字段更完整、置信度更高的一条
- 保留现有初始化、增量、周/月聚合与最终画像生成链路，同时剔除 workflow 接入后已无效或低价值的旧逻辑
- 为后续继续收紧 fallback 标记、增强后台质量诊断和优化工作流输出稳定性打下基础

plugins/member_context/digest_service.py

@@ -5,12 +5,12 @@ from collections import defaultdict
 from datetime import datetime, timedelta
 from typing import Dict, List, Optional, Tuple
 
-import requests
 from loguru import logger
 
 from db.contacts_db import ContactsDBOperator
 from db.member_digest_db import MemberDigestDBOperator
 from db.message_storage import MessageStorageDB
+from plugins.member_context.dify_client import DifyClient
 from plugins.member_context.prompt_builder import MemberContextPromptBuilder
 from utils.compress_chat_data import compress_chat_data
 
@@ -28,12 +28,13 @@ class MemberDigestService:
 
         api_config = self.plugin_config.get("api", {})
         profile_config = self.plugin_config.get("profile", {})
+        self.dify_client = DifyClient(api_config)
 
-        self.ai_enabled = bool(api_config.get("enable", api_config.get("enabled", False)))
-        self.ai_base_url = (api_config.get("base_url") or "").rstrip("/")
-        self.ai_api_key = api_config.get("api_key", "")
-        self.ai_endpoint = str(api_config.get("endpoint", "completion-messages")).lstrip("/")
-        self.ai_timeout = int(api_config.get("request_timeout", 60))
+        self.ai_enabled = self.dify_client.enabled
+        self.ai_base_url = self.dify_client.base_url
+        self.ai_api_key = self.dify_client.api_key
+        self.ai_endpoint = self.dify_client.endpoint
+        self.ai_timeout = self.dify_client.timeout
 
         self.bootstrap_days = int(profile_config.get("bootstrap_days", 365))
         self.daily_message_limit = int(profile_config.get("daily_message_limit", 120))
@@ -301,36 +302,6 @@ class MemberDigestService:
         )
         return digests
 
-    def _build_daily_digest(self, chatroom_id: str, wxid: str, display_name: str,
-                            digest_date: str, messages: List[Dict]) -> Optional[Dict]:
-        prompt = MemberContextPromptBuilder.build_daily_digest_prompt(
-            chatroom_id, wxid, display_name, digest_date, messages
-        )
-        parsed = self._request_ai_json(prompt, tag=f"daily:{digest_date}", chatroom_id=chatroom_id, wxid=wxid)
-        if not parsed:
-            parsed = self._build_daily_digest_fallback(messages)
-        if not parsed:
-            return None
-
-        period_start = f"{digest_date} 00:00:00"
-        period_end = f"{digest_date} 23:59:59"
-        return {
-            "chatroom_id": chatroom_id,
-            "wxid": wxid,
-            "digest_type": "daily",
-            "period_key": digest_date,
-            "period_start": period_start,
-            "period_end": period_end,
-            "display_name": display_name,
-            "source_count": len(messages),
-            "summary_text": parsed.get("summary_text", ""),
-            "structured": parsed,
-            "meta": {
-                "source_type": "messages",
-                "representative_messages": parsed.get("representative_messages", []),
-            },
-        }
-
     def _build_period_digest(self, digest_type: str, chatroom_id: str, wxid: str, display_name: str,
                              period_key: str, period_start: str, period_end: str,
                              items: List[Dict]) -> Optional[Dict]:
@@ -360,64 +331,44 @@ class MemberDigestService:
         }
 
     def _request_ai_json(self, prompt: str, tag: str, chatroom_id: str, wxid: str) -> Optional[Dict]:
-        if not self.ai_enabled or not self.ai_base_url or not self.ai_api_key:
+        if not self.dify_client.is_available():
             return None
-        headers = {
-            "Authorization": f"Bearer {self.ai_api_key}",
-            "Content-Type": "application/json",
-        }
-        payload = {
-            "inputs": {"query": prompt},
-            "response_mode": "blocking",
-            "user": f"member-digest:{chatroom_id}:{wxid}:{tag}",
-        }
-        url = f"{self.ai_base_url}/{self.ai_endpoint}"
-        try:
-            self.LOG.info(f"[成员交互摘要][AI] 发起摘要请求: group={chatroom_id}, wxid={wxid}, tag={tag}")
-            response = requests.post(url, headers=headers, json=payload, timeout=self.ai_timeout)
-            response.raise_for_status()
-            data = response.json()
-            parsed = self._parse_ai_answer(data.get("answer", ""))
-            if parsed:
-                usage = (data.get("metadata") or {}).get("usage", {}) or {}
-                parsed["ai_usage"] = usage
-            return parsed
-        except Exception as e:
-            self.LOG.warning(f"[成员交互摘要][AI] 摘要请求失败: group={chatroom_id}, wxid={wxid}, tag={tag}, error={e}")
+        response = self.dify_client.run(
+            prompt=prompt,
+            user=f"member-digest:{chatroom_id}:{wxid}:{tag}",
+            inputs={"query": prompt, "chatroom_id": chatroom_id, "wxid": wxid, "tag": tag},
+            tag=tag,
+        )
+        if not response:
             return None
+        parsed = self._parse_ai_answer(response.get("text", ""))
+        if parsed:
+            parsed["ai_usage"] = response.get("usage", {}) or {}
+        return parsed
 
     def _request_group_daily_json(self, chatroom_id: str, digest_date: str,
                                   member_labels: List[str], compressed_chat: str) -> List[Dict]:
-        if not self.ai_enabled or not self.ai_base_url or not self.ai_api_key:
+        if not self.dify_client.is_available():
             return []
         prompt = MemberContextPromptBuilder.build_group_daily_digest_prompt(
             chatroom_id, digest_date, member_labels, compressed_chat
         )
-        headers = {
-            "Authorization": f"Bearer {self.ai_api_key}",
-            "Content-Type": "application/json",
-        }
-        payload = {
-            "inputs": {"query": prompt},
-            "response_mode": "blocking",
-            "user": f"member-digest:{chatroom_id}:group-daily:{digest_date}",
-        }
-        url = f"{self.ai_base_url}/{self.ai_endpoint}"
-        try:
-            self.LOG.info(
-                f"[成员交互摘要][AI] 发起群日批量摘要请求: group={chatroom_id}, "
-                f"date={digest_date}, members={len(member_labels)}"
-            )
-            response = requests.post(url, headers=headers, json=payload, timeout=self.ai_timeout)
-            response.raise_for_status()
-            data = response.json()
-            parsed = self._parse_group_daily_answer(data.get("answer", ""))
-            return parsed
-        except Exception as e:
-            self.LOG.warning(
-                f"[成员交互摘要][AI] 群日批量摘要失败: group={chatroom_id}, date={digest_date}, error={e}"
-            )
+        response = self.dify_client.run(
+            prompt=prompt,
+            user=f"member-digest:{chatroom_id}:group-daily:{digest_date}",
+            inputs={
+                "query": prompt,
+                "chatroom_id": chatroom_id,
+                "digest_date": digest_date,
+                "member_labels": "\n".join(member_labels),
+                "compressed_chat": compressed_chat,
+            },
+            tag=f"group-daily:{digest_date}",
+        )
+        if not response:
             return []
+        parsed = self._parse_group_daily_answer(response.get("text", ""))
+        return parsed
 
     def _parse_ai_answer(self, answer: str) -> Optional[Dict]:
         if not answer:
@@ -454,14 +405,38 @@ class MemberDigestService:
         members = parsed.get("members", [])
         if not isinstance(members, list):
             return []
-        normalized = []
+        normalized_map = {}
         for item in members:
             if not isinstance(item, dict):
                 continue
             normalized_item = self._normalize_profile_item(item)
-            if normalized_item.get("wxid"):
-                normalized.append(normalized_item)
-        return normalized
+            wxid = normalized_item.get("wxid")
+            if not wxid:
+                continue
+            existing = normalized_map.get(wxid)
+            if not existing or self._score_profile_item(normalized_item) > self._score_profile_item(existing):
+                normalized_map[wxid] = normalized_item
+        return list(normalized_map.values())
+
+    @staticmethod
+    def _score_profile_item(item: Dict) -> float:
+        if not item:
+            return 0.0
+        score = 0.0
+        for key, value in item.items():
+            if key in {"wxid", "display_name"}:
+                continue
+            if isinstance(value, list):
+                score += len([v for v in value if str(v).strip()]) * 1.0
+            elif isinstance(value, (int, float)):
+                score += float(value)
+            elif str(value).strip():
+                score += 0.8
+        try:
+            score += float(item.get("confidence", 0)) * 2
+        except Exception:
+            pass
+        return score
 
     def _build_daily_digest_fallback(self, messages: List[Dict]) -> Optional[Dict]:
         if not messages: