feat: - 企业家接口进行完善，增加长期记忆、知识库预检、情绪状态机

app/core/viking_client.py

@@ -130,6 +130,47 @@ async def delete_from_tos(object_key: str) -> None:
     await loop.run_in_executor(_executor, _tos_delete_sync, object_key)
 
 
+def _search_knowledge_sync(resource_id: str, query: str, limit: int = 5) -> list:
+    """
+    同步检索 Viking 知识库，返回 PointInfo 列表。
+    在线程池中调用，避免阻塞事件循环。
+    配置：开启重排序（base-multilingual-rerank）、邻近切片扩散1、文本聚合。
+    """
+    from vikingdb.knowledge import VikingKnowledge
+    from vikingdb.auth import APIKey
+    from vikingdb.knowledge.models.search import SearchKnowledgeRequest
+
+    viking_client = VikingKnowledge(
+        host=cfg.VIKING_HOST,
+        region="cn-beijing",
+        auth=APIKey(api_key=cfg.VIKING_API_KEY),
+        scheme="https",
+    )
+    collection = viking_client.collection(resource_id=resource_id)
+    resp = collection.search_knowledge(SearchKnowledgeRequest(
+        query=query,
+        limit=limit,
+        post_processing={
+            "rerank_switch": True,
+            "retrieve_count": max(limit * 3, 15),   # 进入重排的候选数，须 >= limit
+            "chunk_diffusion_count": 1,              # 命中切片上下各扩散1片
+            "chunk_group": True,                     # 按文档原始顺序重排，适合叙述性文档
+        },
+    ))
+    if resp and resp.result and resp.result.result_list:
+        return resp.result.result_list
+    return []
+
+
+async def search_knowledge(resource_id: str, query: str, limit: int = 5) -> list:
+    """
+    异步检索 Viking 知识库，返回 PointInfo 列表。
+    result_list 非空表示知识库有命中，为空则降级到联网搜索。
+    """
+    loop = asyncio.get_event_loop()
+    return await loop.run_in_executor(_executor, _search_knowledge_sync, resource_id, query, limit)
+
+
 async def delete_doc_from_knowledge(resource_id: str, doc_id: str) -> dict:
     """
     调用 Viking /api/knowledge/doc/delete 接口删除知识库文档。

app/db/mongo.py

@@ -438,3 +438,126 @@ def delete_knowledge_doc(doc_id: str) -> bool:
     except Exception as e:
         print(f"MongoDB 知识库文档记录删除失败: {e}")
         return False
+
+
+# ==================== 用户长期记忆 ====================
+
+# facts 使用数组结构 [{key, value, updated_at}]，避免动态字段名带来的 Mongo 注入风险
+user_memory_col = db["user_memory"]
+user_memory_events_col = db["user_memory_events"]
+
+_ALLOWED_FACT_KEYS = {"行业", "市场区域", "职能背景", "痛点", "目标", "产品", "融资阶段"}
+_MAX_FACT_VALUE_LEN = 100
+_MAX_EVENT_LEN = 200
+
+
+def _ensure_user_memory_index():
+    try:
+        user_memory_col.create_index(
+            [("user_id", 1), ("group_id", 1), ("ai_role", 1)], unique=True
+        )
+        user_memory_events_col.create_index(
+            [("user_id", 1), ("group_id", 1), ("timestamp", -1)]
+        )
+        user_memory_events_col.create_index([("user_id", 1), ("keywords", 1)])
+    except Exception:
+        pass
+
+
+def get_user_memory(user_id: str, group_id: str, ai_role: str) -> dict | None:
+    """查询用户在指定群和角色下的记忆文档。"""
+    try:
+        _ensure_user_memory_index()
+        return user_memory_col.find_one(
+            {"user_id": user_id, "group_id": group_id, "ai_role": ai_role},
+            {"_id": 0},
+        )
+    except Exception as e:
+        print(f"MongoDB 查询用户记忆失败: {e}")
+        return None
+
+
+def upsert_user_fact(user_id: str, group_id: str, ai_role: str, key: str, value: str) -> None:
+    """
+    更新或新增用户事实（数组结构），按 key 匹配更新。
+    key 必须在白名单内；value 超长截断。
+    interaction_count 不在此处自增，由调用方统一自增一次。
+    """
+    if key not in _ALLOWED_FACT_KEYS:
+        return
+    value = str(value)[:_MAX_FACT_VALUE_LEN]
+    try:
+        _ensure_user_memory_index()
+        # 先尝试更新已有 key
+        result = user_memory_col.update_one(
+            {"user_id": user_id, "group_id": group_id, "ai_role": ai_role,
+             "facts": {"$elemMatch": {"key": key}}},
+            {"$set": {"facts.$.value": value, "facts.$.updated_at": datetime.now(),
+                      "last_active": datetime.now()}},
+        )
+        if result.matched_count == 0:
+            # key 不存在，追加到数组
+            user_memory_col.update_one(
+                {"user_id": user_id, "group_id": group_id, "ai_role": ai_role},
+                {
+                    "$push": {"facts": {"key": key, "value": value, "updated_at": datetime.now()}},
+                    "$set": {"last_active": datetime.now()},
+                    "$setOnInsert": {"created_at": datetime.now(), "interaction_count": 0},
+                },
+                upsert=True,
+            )
+    except Exception as e:
+        print(f"MongoDB 更新用户事实失败（key={key}）: {e}")
+
+
+def increment_interaction_count(user_id: str, group_id: str, ai_role: str) -> None:
+    """每次 _extract_memory 调用成功后，统一自增一次互动次数。"""
+    try:
+        user_memory_col.update_one(
+            {"user_id": user_id, "group_id": group_id, "ai_role": ai_role},
+            {
+                "$inc": {"interaction_count": 1},
+                "$set": {"last_active": datetime.now()},
+                "$setOnInsert": {"created_at": datetime.now(), "facts": []},
+            },
+            upsert=True,
+        )
+    except Exception as e:
+        print(f"MongoDB 自增互动次数失败: {e}")
+
+
+def append_memory_event(
+    user_id: str, group_id: str, ai_role: str,
+    event: str, keywords: list, trigger_msg_id: str | None = None,
+) -> None:
+    """追加场景记忆事件。"""
+    event = str(event)[:_MAX_EVENT_LEN]
+    try:
+        _ensure_user_memory_index()
+        user_memory_events_col.insert_one({
+            "user_id": user_id, "group_id": group_id, "ai_role": ai_role,
+            "event": event,
+            "keywords": [str(k)[:30] for k in keywords[:10]],
+            "trigger_message_id": trigger_msg_id,
+            "timestamp": datetime.now(),
+        })
+    except Exception as e:
+        print(f"MongoDB 追加记忆事件失败: {e}")
+
+
+def search_memory_events(
+    user_id: str, group_id: str, keywords: list, limit: int = 5,
+) -> list:
+    """按关键词检索近期场景记忆事件，按时间倒序返回。"""
+    if not keywords:
+        return []
+    try:
+        docs = user_memory_events_col.find(
+            {"user_id": user_id, "group_id": group_id,
+             "keywords": {"$in": [str(k)[:30] for k in keywords[:10]]}},
+            {"_id": 0, "event": 1, "timestamp": 1},
+        ).sort("timestamp", -1).limit(limit)
+        return list(docs)
+    except Exception as e:
+        print(f"MongoDB 检索记忆事件失败: {e}")
+        return []

app/db/redis_client.py

@@ -362,6 +362,124 @@ def cache_group_context(group_id: str, context: list, ttl: int = 120):
         print(f"Redis缓存群上下文失败: {e}")
 
 
+# ==================== 情绪状态机 ====================
+
+# 群情绪标签及其数值权重（正值=正面，负值=负面）
+_MOOD_SCORES = {"excited": 2.0, "calm": 0.0, "low": -1.5, "tense": -2.0}
+
+# 各情绪档位对应的语气指令
+_MOOD_TONE = {
+    "excited": "群里气氛热烈，回复可以带点感染力，用“确实”“厉害”接住情绪再展开。",
+    "calm": "保持你一贯的沉稳企业家语气。",
+    "low": "群里气氛偏低落，语气放缓，先共情再给建议，不要上来就分析。",
+    "tense": "群里情绪紧张或焦虑，回复要稳，给确定性而非更多问题，避免追问施压。",
+}
+
+# 滑动窗口最近5条，权重从旧到新（越新权重越高），归一化后使用
+_MOOD_WEIGHTS_BASE = [0.10, 0.10, 0.15, 0.25, 0.40]
+
+
+def classify_mood(text: str) -> tuple:
+    """
+    规则法快速情绪分类，零延迟、零调用成本。
+    返回 (mood: str, intensity: float)
+    """
+    if any(w in text for w in ["太难了", "撑不住", "不想干", "亏了", "失败", "崩了", "没希望"]):
+        return "low", 0.7
+    if any(w in text for w in ["焦虑", "慌", "急", "来不及", "压力"]) or text.count("？") >= 3:
+        return "tense", 0.6
+    if any(w in text for w in ["牛", "太好了", "成了", "恭喜", "涨了", "爆了"]) or "！" in text:
+        return "excited", 0.6
+    return "calm", 0.3
+
+
+def update_group_mood(group_id: str, user_id: str, text: str) -> tuple:
+    """
+    用新消息更新群情绪状态，写入 Redis（TTL 30 分钟）。
+    每条消息以 user_id+text[:20] 作简单去重，避免相同历史消息重复计入。
+    返回 (group_mood: str, intensity: float)
+    """
+    if not redis_client:
+        return "calm", 0.0
+
+    key = f"mood:group:{group_id}"
+    dedup_key = f"{user_id}:{text[:20]}"
+
+    try:
+        raw = redis_client.get(key)
+        state = json.loads(raw) if raw else {"recent_signals": [], "seen_keys": []}
+
+        # 去重：同一消息不重复写入
+        seen_keys = state.get("seen_keys", [])
+        if dedup_key in seen_keys:
+            # 消息已处理，直接返回当前状态
+            return state.get("group_mood", "calm"), state.get("intensity", 0.0)
+
+        mood, _ = classify_mood(text)
+        signals = state.get("recent_signals", [])
+        signals.append({"mood": mood})
+        signals = signals[-5:]  # 保留最近5条
+        seen_keys.append(dedup_key)
+        seen_keys = seen_keys[-50:]  # 去重集合最多50条
+
+        # 取与信号数量对应的尾部权重，然后归一化到和为1
+        weights_raw = _MOOD_WEIGHTS_BASE[-len(signals):]
+        total_w = sum(weights_raw)
+        weights = [w / total_w for w in weights_raw]
+
+        scores = [_MOOD_SCORES[s["mood"]] for s in signals]
+        avg = sum(w * s for w, s in zip(weights, scores))
+
+        if avg >= 1.0:
+            group_mood = "excited"
+        elif avg >= -0.5:
+            group_mood = "calm"
+        elif avg >= -1.5:
+            group_mood = "low"
+        else:
+            group_mood = "tense"
+
+        intensity = round(min(abs(avg) / 2.0, 1.0), 2)
+
+        new_state = {
+            "group_mood": group_mood,
+            "intensity": intensity,
+            "recent_signals": signals,
+            "seen_keys": seen_keys,
+            "updated_at": time.time(),
+        }
+        redis_client.setex(key, 1800, json.dumps(new_state, ensure_ascii=False))
+        return group_mood, intensity
+
+    except Exception as e:
+        print(f"Redis 更新群情绪失败: {e}")
+        return "calm", 0.0
+
+
+def get_group_mood(group_id: str) -> tuple:
+    """读取群当前情绪状态，Redis 不可用时回退到 calm。"""
+    if not redis_client:
+        return "calm", 0.0
+    try:
+        raw = redis_client.get(f"mood:group:{group_id}")
+        if not raw:
+            return "calm", 0.0
+        state = json.loads(raw)
+        return state.get("group_mood", "calm"), state.get("intensity", 0.0)
+    except Exception as e:
+        print(f"Redis 读取群情绪失败: {e}")
+        return "calm", 0.0
+
+
+def build_mood_block(group_id: str) -> str:
+    """构造注入 system prompt 的群氛围说明段。"""
+    mood, intensity = get_group_mood(group_id)
+    if mood == "calm" and intensity < 0.1:
+        return ""  # 平静且强度极低时不注入，减少 prompt 噪声
+    tone = _MOOD_TONE.get(mood, "")
+    return f"\n\n## 当前群氛围\n{mood}（强度 {intensity}）。{tone}"
+
+
 # ==================== 全局实例 ====================
 
 # 频率控制器实例

app/routers/chat_tools.py

@@ -1,3 +1,5 @@
+import asyncio
+import json
 from fastapi import APIRouter, HTTPException
 from datetime import datetime
 
@@ -6,12 +8,14 @@ from ..config.config import Config
 from ..schemas.chat import ChatMessage, ChatResponse, CommentRequest, CirclePromptConfig, HistoricalFigure, \
     RephraseRequest, FigureUpsert, GroupChatRequest, PostCommentBotRequest, RoleBasedChatRequest
 from ..db.souyue_mongo import get_mblog_by_id
-from ..db.mongo import save_chat_log,get_circle_prompt, upsert_circle_prompt, get_all_figures, get_figure_by_id, insert_figure, update_figure, delete_figure, \
-    save_group_ai_reply, get_last_group_response_id
-from ..utils.chat_utils import  get_web_search_tools, get_knowledge_search_tools
+from ..db.mongo import save_chat_log, get_circle_prompt, upsert_circle_prompt, get_all_figures, get_figure_by_id, \
+    insert_figure, update_figure, delete_figure, save_group_ai_reply, get_last_group_response_id, \
+    get_user_memory, upsert_user_fact, append_memory_event, increment_interaction_count
+from ..utils.chat_utils import get_web_search_tools, get_knowledge_search_tools
 from ..db.ai_config import get_config_by_app_name
-from ..db.mongo import  save_chat_history, get_chat_history, delete_chat_history, get_sessions, get_last_response_id
-from ..db.redis_client import get_cached_group_context, cache_group_context
+from ..db.mongo import save_chat_history, get_chat_history, delete_chat_history, get_sessions, get_last_response_id
+from ..db.redis_client import get_cached_group_context, cache_group_context, update_group_mood, build_mood_block
+from ..core.viking_client import search_knowledge
 router = APIRouter()
 
 import re
@@ -70,6 +74,63 @@ def _should_block_role_message(message: str) -> bool:
     return any(keyword in text for group in groups for keyword in group)
 
 
+async def _extract_memory(
+    user_id: str, group_id: str, ai_role: str,
+    trigger_msg: str, ai_reply: str, app_name: str,
+) -> None:
+    """
+    回复成功后异步抽取用户记忆，fire-and-forget。
+    用小模型从对话中提取结构化事实和关键事件，失败仅打印日志不影响主流程。
+    interaction_count 在此处统一自增一次（不在每条 fact 上重复自增）。
+    """
+    try:
+        extract_prompt = (
+            f"从以下群聊交互中抽取值得长期记住的用户事实和关键事件。\n"
+            f"用户消息：{trigger_msg}\nAI回复：{ai_reply}\n"
+            f"只从用户明确陈述的内容中抽取，不要推断或补充。\n"
+            f"可识别的事实类型：行业、市场区域、职能背景、痛点、目标、产品、融资阶段。\n"
+            f"输出严格JSON：{{\"facts\":{{\"行业\":\"\",\"市场区域\":\"\"}},\"events\":[\"事件描述\"],\"keywords\":[\"关键词\"]}}。\n"
+            f"无有价值信息则输出：{{\"facts\":{{}},\"events\":[],\"keywords\":[]}}"
+        )
+        client = get_client(app_name)
+        resp = client.responses.create(
+            model="doubao-seed-2-1-lite-260915",
+            input=[{"role": "user", "content": extract_prompt}],
+            stream=False,
+            store=False,
+            text={"format": {"type": "json_object"}},
+        )
+        raw_text = ""
+        for item in resp.output:
+            if hasattr(item, "type") and item.type == "message" and hasattr(item, "content"):
+                if isinstance(item.content, list):
+                    for c in item.content:
+                        if hasattr(c, "text"):
+                            raw_text += c.text
+                else:
+                    raw_text += str(item.content)
+
+        if not raw_text.strip():
+            return
+        data = json.loads(raw_text)
+
+        # 写入事实（每条 fact 只 upsert，不自增计数）
+        for k, v in data.get("facts", {}).items():
+            if v:
+                upsert_user_fact(user_id, group_id, ai_role, k, str(v))
+
+        # 写入场景事件
+        for ev in data.get("events", []):
+            if ev:
+                append_memory_event(user_id, group_id, ai_role, ev, data.get("keywords", []))
+
+        # 整次 _extract_memory 调用只自增一次互动次数
+        increment_interaction_count(user_id, group_id, ai_role)
+
+    except Exception as e:
+        print(f"记忆抽取失败（user={user_id}）: {e}")
+
+
 def _build_prompt(product_text: str, prompt_config: dict) -> str:
     name = prompt_config.get("name", "兴趣圈")
     role = prompt_config.get("role", "活跃用户")
@@ -550,15 +611,16 @@ async def entrepreneur_chat(request: RoleBasedChatRequest):
                     if group_context:
                         cache_group_context(group_id, group_context, ttl=120)
 
-                # 构造多模态消息格式
+                # 构造多模态消息格式，同步更新群情绪状态（去重由 update_group_mood 内部处理）
                 if group_context:
                     for msg in group_context[-20:]:  # 最近20条
-                        user_id = msg['user_id']
+                        msg_user_id = msg['user_id']
                         content_parts = []
 
-                        # 添加文本内容
+                        # 添加文本内容，并更新群情绪
                         if msg.get('content'):
-                            content_parts.append({"type": "input_text", "text": f"{user_id}: {msg['content']}"})
+                            content_parts.append({"type": "input_text", "text": f"{msg_user_id}: {msg['content']}"})
+                            update_group_mood(group_id, msg_user_id, msg['content'])
 
                         # 添加图片
                         for img_url in msg.get('images', []):
@@ -651,6 +713,24 @@ async def entrepreneur_chat(request: RoleBasedChatRequest):
 如果不需要回应，保持沉默（不输出任何内容）。
 """
 
+        # 【v3.0新增】注入用户长期记忆（已确认的事实背景）
+        mem = get_user_memory(request.user_id, group_id, "entrepreneur")
+        if mem and mem.get("facts"):
+            fact_lines = [
+                f"- {f['key']}: {f['value']}"
+                for f in mem["facts"]
+                if isinstance(f, dict) and f.get("key") and f.get("value")
+            ]
+            if fact_lines:
+                system_prompt += (
+                    "\n\n## 你对当前发言用户的已知背景\n"
+                    + "\n".join(fact_lines)
+                    + "\n参照这些背景给出更具体的建议，但不要生硬复述。"
+                )
+
+        # 【v3.0新增】注入群情绪状态
+        system_prompt += build_mood_block(group_id)
+
         # 构造请求消息
         system_msg = {"role": "system", "content": [{"type": "input_text", "text": system_prompt}]}
 
@@ -669,8 +749,37 @@ async def entrepreneur_chat(request: RoleBasedChatRequest):
         else:
             api_messages = [system_msg, {"role": "user", "content": user_content}]
 
-        # 配置工具：联网搜索 + 知识库
-        tools = get_knowledge_search_tools(knowledge_id,request.app_name)
+        # 【v3.0新增】知识库预检：直接调用 Viking SDK 检索
+        # 有命中 → 将切片内容作为上下文注入，不挂 Ark 知识库工具
+        # 无命中 → 挂联网搜索工具
+        tools = []
+        if knowledge_id:
+            try:
+                kb_results = await search_knowledge(
+                    resource_id=knowledge_id,
+                    query=request.message,
+                    limit=5,
+                )
+            except Exception as e:
+                print(f"知识库预检失败，降级到联网搜索: {e}")
+                kb_results = []
+        else:
+            kb_results = []
+
+        if kb_results:
+            # 提取切片内容，拼接成上下文段落
+            kb_texts = [r.content for r in kb_results if r.content]
+            if kb_texts:
+                kb_context_block = [{"type": "input_text", "text": "【知识库参考资料】\n" + "\n\n".join(kb_texts) + "\n\n请优先基于以上资料回答，资料中没有的内容不要编造。"}]
+                # 将知识库上下文插入到 user content 前面
+                if context_messages:
+                    api_messages[-1]["content"] = kb_context_block + api_messages[-1]["content"]
+                else:
+                    api_messages[-1]["content"] = kb_context_block + user_content
+            # 有知识库命中时不挂任何工具
+        else:
+            # 无命中，降级到联网搜索
+            tools = get_web_search_tools()
 
         # 调用 AI 客户端
         response = client.responses.create(
@@ -710,6 +819,17 @@ async def entrepreneur_chat(request: RoleBasedChatRequest):
                 app_name=request.app_name,
             )
 
+        # 【v3.0新增】异步抽取用户记忆（fire-and-forget，失败不影响主流程）
+        if result:
+            asyncio.create_task(_extract_memory(
+                user_id=request.user_id,
+                group_id=group_id,
+                ai_role="entrepreneur",
+                trigger_msg=request.message,
+                ai_reply=result,
+                app_name=request.app_name,
+            ))
+
         if not result:
             save_chat_log(
                 user_id=request.user_id,

requirements.txt

@@ -13,7 +13,8 @@ httpx==0.28.1
 openai==2.30.0
 tls-sig-api-v2==1.1
 python-multipart==0.0.32
-websockets>=12.0
-tos>=2.7.0
+websockets==16.0
+tos==2.9.3
+vikingdb-python-sdk==0.1.34
 #llama-index-core==0.14.25
 #Jinja2==3.1.6