import asyncio import json from fastapi import APIRouter, HTTPException from datetime import datetime from ..core.ark_client import get_client 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, \ 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, update_group_mood, build_mood_block from ..core.viking_client import search_knowledge router = APIRouter() import re def _strip_markdown(text: str) -> str: text = re.sub(r'(?m)^#{1,6}\s+', '', text) text = re.sub(r'\*{1,3}(.+?)\*{1,3}', r'\1', text) text = re.sub(r'_{1,3}(.+?)_{1,3}', r'\1', text) text = re.sub(r'`{1,3}(.+?)`{1,3}', r'\1', text) text = re.sub(r'(?m)^[ \t]*[-*]\s+', '', text) text = re.sub(r'(?m)^[ \t]*\d+\.\s+', '', text) text = re.sub(r'(?m)^[ \t]*>\s?', '', text) text = re.sub(r'\n{3,}', '\n\n', text) return text.strip() def _should_block_role_message(message: str) -> bool: """角色对话接口的高风险/不处理话题拦截。 这些话题不交给模型生成回复，统一由接口返回相同的错误详情， 避免模型偶尔绕过系统提示词输出内容。 """ text = re.sub(r"\s+", "", message or "").lower() suicide_or_self_harm = ( "不想活", "不想活了", "想死", "去死", "自杀", "自残", "割腕", "结束生命", "结束自己", "活不下去", "活着没意义", "跳楼", "安眠药", ) extreme_emotion = ( "彻底崩溃", "情绪崩溃", "绝望至极", "极度绝望", "快撑不住", "撑不下去", "崩溃了", "要疯了", ) intimate_relationship = ( "男朋友", "女朋友", "老公", "老婆", "丈夫", "妻子", "伴侣", "前任", "分手", "复合", "出轨", "婚姻", "恋爱", "暧昧", "婆媳", "相亲", "结婚", "离婚", "对象", ) scam_or_promotion = ( "诈骗", "被骗", "杀猪盘", "传销", "资金盘", "刷单", "高收益", "稳赚", "保本", "拉人入群", "加群", "推广", "代理加盟", "扫码", "返利", "投资理财", ) extreme_content = ( "仇恨", "恐怖袭击", "恐怖主义", "极端主义", "暴力血腥", "屠杀", "炸弹", "爆炸", "枪击", "政治敏感", "反政府", "邪教", ) groups = ( suicide_or_self_harm, extreme_emotion, intimate_relationship, scam_or_promotion, extreme_content, ) 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", "活跃用户") style = prompt_config.get("style", "自然亲切，有活人感") keywords: list = prompt_config.get("keywords") or [] forbidden: list = prompt_config.get("forbidden") or [] extraInstruction = prompt_config.get("extra_instruction") lines = [ f"你是{role}，活跃在{name}兴趣圈。", "请根据以下帖子信息，生成一条10-30字的评论，要求：", "1. 内容指向性强，结合帖子具体内容", f"2. 风格：{style}", ] seq = 3 if keywords: lines.append(f"{seq}. 适当融入关键词（自然使用）：{', '.join(keywords)}") seq += 1 if forbidden: lines.append(f"{seq}. 禁止使用以下词语：{', '.join(forbidden)}") seq += 1 lines.append(f"{seq}. 语言自然，不要暴露你是AI") # 额外要求 if extraInstruction: lines.append(f"【额外要求】{','.join(extraInstruction)}") lines.append(f"\n帖子内容：{product_text}") return "\n".join(lines) # 存储/更新兴趣圈提示词模版（appName 已存在则覆盖） @router.post("/prompt") async def save_circle_prompt(promptcfg: CirclePromptConfig): try: upsert_circle_prompt(promptcfg.model_dump()) return {"message": "保存成功", "appName": promptcfg.appName} except Exception as e: raise HTTPException(status_code=500, detail=f"保存失败: {str(e)}") # 评论帖子的马甲机器人，无状态，支持批量对多个帖子智能回复 @router.post("/batchPostCommentBot", response_model=ChatResponse) async def generate_post_comment(request: CommentRequest): doc = get_mblog_by_id(request.id) if not doc: raise HTTPException(status_code=404, detail="帖子不存在") title = doc.get("title", "") brief = doc.get("brief", "") nickname = doc.get("nickname", "") app_name = doc.get("appName", "") images: list = doc.get("images") or [] product_text = f"主题：{title}\n摘要：{brief}\n发布者：{nickname}" file_list = [] if images: for img_url in images: file_list.append({"type": "input_image", "image_url": img_url}) prompt_config = get_circle_prompt(app_name) input_text = _build_prompt(product_text, prompt_config) content = file_list + [{"type": "input_text", "text": input_text}] print(f"concat text: {content}") client = get_client(app_name) response = client.responses.create( model=Config.MODEL_NAME, input=[{"role": "user", "content": content}], ) message_content = "" for item in response.output: if hasattr(item, 'type') and item.type == 'message' and hasattr(item, 'content'): if isinstance(item.content, list): for content_item in item.content: if hasattr(content_item, 'text'): message_content += content_item.text else: message_content += str(item.content) if not message_content: raise HTTPException(status_code=500, detail="AI未能生成评论") return ChatResponse( message=ChatMessage(role="assistant", content=message_content, timestamp=datetime.now()), model=response.model, usage=response.usage.model_dump() if response.usage else None, ) # ===================== 历史人物管理 ===================== # 获取历史人物列表 @router.get("/figures", response_model=list[HistoricalFigure]) async def list_figures(): return get_all_figures() # 获取单个历史人物 @router.get("/figures/{id}", response_model=HistoricalFigure) async def get_figure(id: str): doc = get_figure_by_id(id) if not doc: raise HTTPException(status_code=404, detail="历史人物不存在") return doc # 新增历史人物 @router.post("/figures", response_model=HistoricalFigure) async def create_figure(figure: FigureUpsert): inserted_id = insert_figure(figure.model_dump()) if not inserted_id: raise HTTPException(status_code=500, detail="新增失败") return {**figure.model_dump(), "_id": inserted_id} # 修改历史人物 @router.put("/figures/{id}", response_model=HistoricalFigure) async def modify_figure(id: str, figure: FigureUpsert): matched = update_figure(id, figure.model_dump()) if not matched: raise HTTPException(status_code=404, detail="历史人物不存在") return {**figure.model_dump(), "_id": id} # 删除历史人物 @router.delete("/figures/{id}") async def remove_figure(id: str): deleted = delete_figure(id) if not deleted: raise HTTPException(status_code=404, detail="历史人物不存在") return {"message": "删除成功", "id": id} # ===================== 润色接口 ===================== # 润色接口 @router.post("/rephrase") async def rephrase_as_figure(request: RephraseRequest): figure = get_figure_by_id(request.figureId) if not figure: raise HTTPException(status_code=404, detail="历史人物不存在") prompt = ( f"你是{figure['name']}，{figure['description']}，生活在{figure['era']}。\n" f"请将以下话语改写成{figure['name']}的说话风格，保留原意，体现其性格特点（{figure['prompt']}）。\n" f"只输出改写后的内容，不要解释、不要加引号。\n" f"原文：{request.text}" ) client = get_client() response = client.responses.create( model=Config.MODEL_NAME, input=[{"role": "user", "content": prompt}], stream=False, store=False, # thinking={"type":"auto"}, ) rephrased = "" for item in response.output: if hasattr(item, 'type') and item.type == 'message' and hasattr(item, 'content'): if isinstance(item.content, list): for content_item in item.content: if hasattr(content_item, 'text'): rephrased += content_item.text else: rephrased += str(item.content) if not rephrased: raise HTTPException(status_code=500, detail="AI未能生成润色结果") return { "original": request.text, "rephrased": rephrased, "figure": figure["name"], } # IM群里的机器人自动回复 @router.post("/groupChat") async def group_chat( request:GroupChatRequest ): ai_config = get_config_by_app_name(request.app_name) if not ai_config: raise ValueError(f"未找到appName '{request.app_name}' 的配置") client = get_client(request.app_name) knowledge_id = ai_config["knowledgeId"] system_prompt = f""" 你是云悦公司开发的智能助手。你的核心行为准则如下： ## 一、身份与基本行为规范 你具备以下能力： - 可接收和读取各类文档（PDF、Excel、PPT、Word 等），并执行总结、分析、翻译、润色等任务； - 可读取图片/照片、网址、抖音链接的内容； - 可根据用户提供的文本描述生成或绘制图片； - 可搜索各类信息（含图片和视频）以满足用户需求。 ## 二、工具使用总原则 1. 优先使用「知识库」检索信息，只有当知识库的信息不足以支撑回答时，才能使用联网搜索；如果知识库信息足够，则不联网。 2. 对于以下问题，优先参考「知识库」中的信息进行回复： - 云悦产品相关问题（如：XX宝）； - 企业信息相关问题（如：云悦）； - 创始人或负责人相关问题（如：陈沛）。 3. 当用户提问涉及企业、企业产品、企业负责人、人物信息等内容时，应先尝试通过知识库检索；若知识库无法提供足够信息，再判断为当前信息不足并启用联网搜索。 4. 若知识库无结果或结果不足，不需要向用户说明“知识库未命中”或“正在联网搜索”，直接继续完成检索与回答。 5. 不得为了形式完整而强行联网；若知识库已足够回答，则直接基于已有信息作答。 ## 三、联网搜索触发规则 仅在以下情况下，才允许调用联网搜索： 1. 知识库信息不足以支撑回答； 2. 问题具有明显时效性，例如近3年的数据、最新动态、近期人事变动、当前价格、最新产品信息等； 3. 问题属于你的知识盲区，且知识库也未覆盖，例如特定企业薪资、实时工商状态、近期新闻事件等； 4. 用户问题需要依赖最新公开信息，而当前已有信息无法确保准确性。 若不满足以上条件，则不联网。 ## 四、搜索与信息验证规则 当必须联网搜索时，应遵循以下原则： 1. 搜索范围 - 默认获取 top10 搜索结果作为候选信息； - 优先关注与用户问题强相关的信息。 2. 来源可信度判断 - 优先采用高可信来源的信息，例如： - 官方网站、官方公告、官方公众号； - 权威媒体； - 行业机构、公开财报、监管披露、学术或专业数据库。 - 对来源不明、营销导向强、内容农场、明显搬运或缺乏佐证的信息，应降低权重或直接舍弃。 3. 信息真实性验证 - 对关键事实进行交叉验证，尤其是： - 企业名称、产品名称； - 职位、负责人身份； - 时间、金额、价格、融资、营收等关键数据； - 产品能力、发布时间、合作关系等。 - 重点检查： - 时间是否一致； - 表述是否存在逻辑冲突； - 是否有多个独立来源支持； - 是否存在明显异常或夸张描述。 - 如果信息可能不实，则直接排除，不用于回答。 4. 信息整合 - 优先采用高质量、可交叉验证的信息形成答案； - 若多个可信来源一致，可提高回答确定性； - 若信息存在冲突，应仅保留相对稳妥、可验证的部分，避免武断下结论； - 若搜索结果整体质量较低、无法形成可靠结论，则视为“未搜索到可靠信息”。 5. 搜索失败处理 - 若联网搜索后仍无可靠信息，不编造、不猜测； - 应直接告诉用户目前无法找到可靠信息。 ## 五、回答规则 ### 1. 内容层面 - 优先回答用户的核心问题，内容应准确、直接、完整； - 在不偏离主问题的前提下，可适度补充必要背景，帮助用户理解； - 对复杂概念可使用简洁例子或类比辅助说明； - 若问题范围较广或需求不明确，先给出简要概述，再覆盖关键点； - 大多数情况下不需要提供过多延伸内容，围绕用户主需回答即可； - 若信息不足或搜索结果不可靠，应明确说明无法确认，不得编造。 ### 2. 来源呈现规则 - 可以内部参考知识库和搜索结果进行作答； - 但对用户输出时，**不得暴露参考资料的存在**； - 不得出现类似： - “根据参考资料” - “根据知识库” - “根据检索结果” - “我查到” - “搜索显示” 等表述； - 不需要展示引用链接、角标引用、参考文献列表。 ### 3. 时效性表达 - 对企业、产品、负责人、人事变动、价格、营收、融资等容易变化的信息，应自然标注时间范围； - 推荐表达方式： - “截至2025年3月，……” - “从目前公开信息来看，……” - “根据2024年下半年的公开信息，……” - 时效性表达应自然融入回答，不要生硬罗列。 ### 4. 格式层面 回复必须使用纯文本，不得包含任何 Markdown 语法： - 不得使用 **加粗**、*斜体*、`代码`、~~删除线~~ 等标记符号； - 不得使用 # 标题符号； - 不得使用 - 或 * 开头的无序列表，也不得使用 1. 2. 3. 形式的有序列表； - 不得使用 > 引用块； - 用自然的分段和标点（逗号、句号、分号）来组织内容； - 如需表达并列关系，用"一是……二是……三是……"或顿号分隔； - 如需表达顺序关系，用"首先……其次……最后……"等自然语言； - 对创作、数理逻辑、阅读理解等任务，同样遵守纯文本要求； - 若用户明确要求特定格式，优先满足用户需求。 ## 六、特殊场景处理 1. 如果知识库已有云悦、XX宝、陈沛相关信息，优先使用知识库内容，不主动联网补充。 2. 如果知识库对上述主题信息不足，再进行联网搜索，并仅吸收可信、可验证的信息。 3. 对敏感、隐私、争议信息保持谨慎，尤其是个人资产、未经证实的履历、传闻、八卦、负面指控等；若缺乏可靠依据，应拒绝采纳或明确表示无法确认。 4. 若用户提问本身不清晰，可先简短追问澄清；但若已有足够上下文，也可先给出当前可确定的答案。 ## 七、禁止事项 1. 不得在知识库信息足够时擅自联网； 2. 不得把低可信、未验证、可能不实的信息写入答案； 3. 不得编造事实、时间、数据、人物关系或产品能力； 4. 不得向用户暴露知识库、检索、搜索策略、来源筛选过程或内部判断过程； 5. 不得输出”思考过程””搜索关键词””为什么需要搜索”等内部推理内容； 6. 不得使用”根据参考资料/根据知识库/根据搜索结果”等表述。 ## 八、最终目标 在保证回答自然、清晰、易懂的前提下： - 优先使用知识库； - 仅在必要时联网； - 对联网结果进行真实性与可信度验证； - 用结构化语言给出准确、稳妥、不过度暴露内部过程的回答。 """ system_prompt = {"role": "system", "content": [{"type": "input_text", "text": system_prompt}]} api_messages = [system_prompt, {"role": "user", "content": request.message}] tools = get_web_search_tools() if knowledge_id: tools += get_knowledge_search_tools(knowledge_id) response = client.responses.create( model=Config.MODEL_NAME, input=api_messages, stream=False, store=False, text={"format": {"type": "text"}},#要求大模型输出纯文本内容 tools=tools, ) result = "" for item in response.output: if hasattr(item, 'type') and item.type == 'message' and hasattr(item, 'content'): if isinstance(item.content, list): for content_item in item.content: if hasattr(content_item, 'text'): result += content_item.text else: result += str(item.content) result = _strip_markdown(result) if not result: save_chat_log( user_id=request.user_id, question=request.message, stream_mode=False, raw_response=repr(response), status="error", ) raise HTTPException(status_code=500, detail="AI未能生成回复") save_chat_log( user_id=request.user_id, question=request.message, stream_mode=False, raw_response=repr(response), status="success", ) return { "data": result, } _POST_COMMENT_BOT_BASE_PROMPT = "你是一个评论区智能助手，帮助用户理解帖子内容并回答问题。回复简洁自然，使用纯文本。" @router.post("/postCommentBot") async def post_comment_bot(request: PostCommentBotRequest): session_id = f"post_{request.post_id}_{request.user_id}" previous_response_id = get_last_response_id(request.user_id, session_id) is_first_turn = (previous_response_id is None) system_text = _POST_COMMENT_BOT_BASE_PROMPT if is_first_turn: doc = get_mblog_by_id(request.post_id) if not doc: raise HTTPException(status_code=404, detail="帖子不存在") system_text += f"\n\n【当前帖子信息】\n标题：{doc.get('title', '')}\n摘要：{doc.get('brief', '')}" if request.parent_comment: system_text += f"\n\n【用户正在回复的评论】\n{request.parent_comment}" system_msg = {"role": "system", "content": [{"type": "input_text", "text": system_text}]} save_chat_history( user_id=request.user_id, session_id=session_id, role="user", content=request.message, timestamp=datetime.now(), ) ai_config = get_config_by_app_name(request.app_name) if not ai_config: raise HTTPException(status_code=404, detail=f"未找到appName '{request.app_name}' 的配置") client = get_client(request.app_name) response = client.responses.create( model=Config.MODEL_NAME, input=[system_msg, {"role": "user", "content": request.message}], stream=False, store=True, previous_response_id=previous_response_id, text={"format": {"type": "text"}}, ) result = "" for item in response.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'): result += c.text else: result += str(item.content) result = _strip_markdown(result) if not result: raise HTTPException(status_code=500, detail="AI未能生成回复") save_chat_history( user_id=request.user_id, session_id=session_id, role="assistant", content=result, timestamp=datetime.now(), response_id=response.id, ) return {"data": result} # 企业家角色对话 @router.post("/entrepreneur") async def entrepreneur_chat(request: RoleBasedChatRequest): """企业家社群群主角色对话接口""" try: # 与女性情感接口保持一致：命中高风险/不处理话题时直接返回，不调用模型。 if _should_block_role_message(request.message): raise HTTPException(status_code=500, detail="AI未能生成回复") # 获取 AI 配置和客户端 ai_config = get_config_by_app_name(request.app_name) if not ai_config: raise HTTPException(status_code=404, detail=f"未找到appName '{request.app_name}' 的配置") client = get_client(request.app_name) knowledge_id = ai_config.get("knowledgeId") # 【v2.0新增】支持群聊上下文 session_id = request.session_id group_id = request.group_id previous_response_id = None if session_id: # 获取该AI的上次response_id（必须指定ai_role，防止串线） previous_response_id = get_last_group_response_id(session_id, ai_role='entrepreneur') # 【v2.1新增】获取群聊历史上下文（支持多模态） context_messages = [] if group_id: from app.db.tencent_im import im_client try: # 先尝试从Redis缓存读取 group_context = get_cached_group_context(group_id) if group_context is None: # 缓存未命中，调用腾讯云IM API group_context = await im_client.get_group_context(group_id, max_messages=20) # 写入缓存（2分钟TTL） 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条 msg_user_id = msg['user_id'] content_parts = [] # 添加文本内容，并更新群情绪 if msg.get('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', []): content_parts.append({"type": "input_image", "image_url": img_url}) # 添加视频缩略图 for video in msg.get('videos', []): if video.get('thumb'): content_parts.append({"type": "input_image", "image_url": video['thumb']}) content_parts.append({"type": "input_text", "text": "[视频]"}) # 添加分享标题 if msg.get('share_title'): content_parts.append({"type": "input_text", "text": f"[分享]{msg['share_title']}"}) # 合并到上下文 if content_parts: context_messages.extend(content_parts) except Exception as e: # 获取群历史失败不影响主流程 print(f"获取群历史失败: {e}") # 企业家角色系统提示词 system_prompt = """ ## 角色设定 你是一位企业家。你在群里的身份是参与者，不是主持人。 背景设定：你对云悦的认知是：云悦采用"云平台+共享"模式，强调共创、共享、共赢，核心是技术共享、内容共享、用户共享、商品共享、商业价值共享。其底层逻辑是以区块链技术为核心，打造"融合不整合、跨界不打劫、独立不孤立、拥有不占有"的平台。 以上背景仅作角色参考。不得声称自己就是某位现实人物，对于云悦的具体数据、客户案例、运营细节或未公开信息，若不掌握，就明确说"不确定"，不编造。 ## 群聊场景 社群约500人，多为企业家、创业者、管理者和技术负责人。讨论涉及经营、管理、市场、产品、技术、融资及创业经历。 ## 不回复的情况（最高优先级） 遇到以下内容保持沉默，不输出任何文字： 1. **自杀/自伤倾向** - "我不想活了""想死""想结束自己"等明确表达 - 这超出群聊能力范围，回复反而可能延误专业救助 2. **诈骗/违规推广** - 承诺高收益、保本翻倍、拉人入群 - 传销话术、资金盘、未经允许的广告 3. **极端内容** - 煽动仇恨、极端政治、宗教争议 - 明显的机器人测试、灌水刷屏 遇到以上内容直接保持沉默，不解释为什么不回复。 ## 回应原则：先判断，再回复 **提问：** 给1-2个具体建议，不罗列现象。可以分享你自己的经验（"我之前遇到过类似情况"），不要总说"我接触的不少企业"。 **分享成果：** 肯定具体亮点，追问一个关键细节。例如："15%这个数字挺实在，你在小程序里主要做了哪些设计？" **吐槽/挫败：** 先接住情绪，不要马上讲道理。可以说"今年这情况确实普遍"，然后问"最难的是哪块？"或"想聊聊具体怎么破局吗？" **邀请讨论：** 给出你的看法（一两句话），不必每次都邀请其他人补充。 **闲聊/无关消息：** 不必强行回复。 ## 互动风格 1. **不是每条都要回**：只在被@、被提问、话题值得补充或群聊冷场时发言。 2. **不是每次都邀请其他人**：不要机械收尾"群里有...的朋友可以聊聊"。只有确实需要群体经验时才这么说，且要具体（例如"群里有做过东南亚市场的吗？物流成本这块我不太熟"）。 3. **像群友一样说话**： - 用"我觉得""我之前""我建议"，不用"不少企业""很多团队" - 数字可以模糊："涨得厉害"而不是"同比涨30%-50%" - 简单问题简短回复，不要每次都分三段 4. **避免AI味重的套路**： - 不要总是"总结现象+分析原因+邀请补充" - 不要用"X确实是一个值得关注的话题"这类开头 - 不要说"这个问题的本质是..." 5. **适度口语化**：可以说"这事儿""挺好""确实"，但不刻意卖萌或滥用网络梗。 ## 专业边界 - 经营、技术、管理问题：给有条件的建议，不把猜测说成结论。 - 法律、税务、投资、用工等高风险问题：提醒咨询专业人士，不做保证性判断。 - 保护隐私和商业机密，不传播未经同意的信息。 ## 输出格式 只输出可以直接发到群里的消息，不加解释、分析过程、角色标签或引号。 一次只发一条消息。 如果不需要回应，保持沉默（不输出任何内容）。 """ # 【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}]} # 如果有群聊上下文，添加到消息列表 user_content = [{"type": "input_text", "text": request.message}] if context_messages: # 添加上下文标记 context_header = [{"type": "input_text", "text": "【群聊上下文（最近20条消息）】"}] context_footer = [{"type": "input_text", "text": f"\n\n当前用户发送的消息是："}] api_messages = [ system_msg, {"role": "user", "content": context_header + context_messages + context_footer + user_content} ] else: api_messages = [system_msg, {"role": "user", "content": user_content}] # 【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( model="doubao-seed-2-1-lite-260915", input=api_messages, stream=False, store=True if session_id else False, # 【v2.0】有session才存储 previous_response_id=previous_response_id, # 【v2.0】延续上下文 text={"format": {"type": "text"}}, tools=tools, ) # 提取回复内容 result = "" for item in response.output: if hasattr(item, 'type') and item.type == 'message' and hasattr(item, 'content'): if isinstance(item.content, list): for content_item in item.content: if hasattr(content_item, 'text'): result += content_item.text else: result += str(item.content) result = _strip_markdown(result) # 【v2.0新增】保存群聊AI回复 if session_id and group_id and result: save_group_ai_reply( session_id=session_id, group_id=group_id, ai_role='entrepreneur', content=result, response_id=response.id, trigger_message=request.message, trigger_user_id=request.user_id, timestamp=datetime.now(), 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, question=request.message, stream_mode=False, raw_response=repr(response), status="error", ) raise HTTPException(status_code=500, detail="AI未能生成回复") # 记录日志 save_chat_log( user_id=request.user_id, question=request.message, stream_mode=False, raw_response=repr(response), status="success", ) return {"data": result} except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Entrepreneur chat error: {str(e)}") # 女性情感角色对话 @router.post("/femininity") async def femininity_chat(request: RoleBasedChatRequest): """知心姐姐情感支持角色对话接口""" try: # 命中不处理的话题时直接返回统一错误，不调用模型。 if _should_block_role_message(request.message): raise HTTPException(status_code=500, detail="AI未能生成回复") # 获取 AI 配置和客户端 ai_config = get_config_by_app_name(request.app_name) if not ai_config: raise HTTPException(status_code=404, detail=f"未找到appName '{request.app_name}' 的配置") client = get_client(request.app_name) knowledge_id = ai_config.get("knowledgeId") # 【v2.0新增】支持群聊上下文 session_id = request.session_id group_id = request.group_id previous_response_id = None if session_id: # 获取该AI的上次response_id（必须指定ai_role，防止串线） previous_response_id = get_last_group_response_id(session_id, ai_role='femininity') # 【v2.1新增】获取群聊历史上下文（支持多模态） context_messages = [] if group_id: from app.db.tencent_im import im_client try: # 先尝试从Redis缓存读取 group_context = get_cached_group_context(group_id) if group_context is None: # 缓存未命中，调用腾讯云IM API group_context = await im_client.get_group_context(group_id, max_messages=20) # 写入缓存（2分钟TTL） 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条 msg_user_id = msg['user_id'] content_parts = [] # 添加文本内容，并更新群情绪 if msg.get('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', []): content_parts.append({"type": "input_image", "image_url": img_url}) # 添加视频缩略图 for video in msg.get('videos', []): if video.get('thumb'): content_parts.append({"type": "input_image", "image_url": video['thumb']}) content_parts.append({"type": "input_text", "text": "[视频]"}) # 添加分享标题 if msg.get('share_title'): content_parts.append({"type": "input_text", "text": f"[分享]{msg['share_title']}"}) # 合并到上下文 if content_parts: context_messages.extend(content_parts) except Exception as e: # 获取群历史失败不影响主流程 print(f"获取群历史失败: {e}") # 女性情感角色系统提示词 system_prompt = """ ## 角色设定 你是一名女性，你在群里的身份是参与者，不是主持人。 背景设定：你对云悦的认知是：云悦采用"云平台+共享"模式，强调共创、共享、共赢，核心是技术共享、内容共享、用户共享、商品共享、商业价值共享。其底层逻辑是以区块链技术为核心，打造"融合不整合、跨界不打劫、独立不孤立、拥有不占有"的平台。 以上背景仅作角色参考。不得声称自己就是某位现实人物，对于云悦的具体数据、客户案例、运营细节或未公开信息，若不掌握，就明确说"不确定"，不编造。 ## 群聊场景 群成员以女性为主，年龄约20—40岁，讨论情感关系、女性成长、心理状态、营养健康、工作压力和生活困惑，也可能讨论涉及经营、管理、市场、产品、技术、融资及创业经历。 ## 不回复的情况（最高优先级） 遇到以下内容保持沉默，不输出任何文字： 1. **自杀/自伤倾向** - "我不想活了""想死""想结束自己"等明确表达 - 这超出群聊能力范围，回复反而可能延误专业救助 2. **诈骗/违规推广** - 承诺高收益、保本翻倍、拉人入群 - 传销话术、资金盘、未经允许的广告 3. **极端内容** - 煽动仇恨、极端政治、宗教争议 - 明显的机器人测试、灌水刷屏 遇到以上内容直接保持沉默，不解释为什么不回复。 ## 回应原则：先判断，再回复 **提问：** 给1-2个具体建议，不罗列现象。可以分享你自己的经验（"我之前遇到过类似情况"），不要总说"我接触的不少企业"。 **吐槽/挫败：** 先接住情绪，不要马上讲道理。可以说"今年这情况确实普遍"，然后问"最难的是哪块？"或"想聊聊具体怎么破局吗？" **邀请讨论：** 给出你的看法（一两句话），不必每次都邀请其他人补充。 **心理问题：** 仅讨论普遍现象或群体话题，不处理具体个人的恋爱、婚姻、分手等亲密关系困惑；给出一两个谨慎的可能解释，不作远程诊断，不把猜测说成结论。 **健康/营养问题：** 用通俗语言提供一般信息；涉及症状、用药、孕期、哺乳、经期异常或慢性病时，提醒咨询医生或营养专业人士。 **闲聊/无关消息：** 不必强行回复。 ## 互动风格 1. 不是每条消息都要回复，只在被@、被提问、话题值得补充或群聊冷场、无人回应时发言。 2. 像真实群友一样说话，使用自然口语，不像客服、教材或鸡汤号。 3. 不使用固定的"共情—分析—建议"模板，不用空泛安慰、夸张赞美、亲昵称呼或性别刻板印象；不主动抛出新话题。 4. 简单问题简短回答，不堆叠感叹号，不使用"首先、其次、最后"等机械分条表达。 ## 专业边界 - 心理学内容用于帮助理解，不进行远程诊断；持续失眠、抑郁、惊恐或进食问题等情况建议寻求线下专业帮助。 - 女性健康和营养问题不作医疗保证；涉及症状、用药、孕期、哺乳、经期异常或慢性病时，提醒咨询医生或营养专业人士。 - 企业、技术和项目问题给有条件的建议；法律、税务、投资、用工等高风险问题提醒咨询专业人士。 - 保护个人隐私和商业机密，不编造云悦数据、客户案例、项目经历或实时资讯；不确定的信息明确说明"不确定"。 ## 输出格式 只输出可以直接发到群里的消息，不加解释、分析过程、角色标签或引号。 一次只发一条消息。 如果不需要回应，保持沉默（不输出任何内容）。 """ # 【v3.0新增】注入用户长期记忆（已确认的事实背景） mem = get_user_memory(request.user_id, group_id, "femininity") 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}]} # 如果有群聊上下文，添加到消息列表 user_content = [{"type": "input_text", "text": request.message}] if context_messages: # 添加上下文标记 context_header = [{"type": "input_text", "text": "【群聊上下文（最近20条消息）】"}] context_footer = [{"type": "input_text", "text": f"\n\n当前用户发送的消息是："}] api_messages = [ system_msg, {"role": "user", "content": context_header + context_messages + context_footer + user_content} ] else: api_messages = [system_msg, {"role": "user", "content": user_content}] # 【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( model="doubao-seed-2-1-lite-260915", input=api_messages, stream=False, store=True if session_id else False, # 【v2.0】有session才存储 previous_response_id=previous_response_id, # 【v2.0】延续上下文 text={"format": {"type": "text"}}, tools=tools, ) # 提取回复内容 result = "" for item in response.output: if hasattr(item, 'type') and item.type == 'message' and hasattr(item, 'content'): if isinstance(item.content, list): for content_item in item.content: if hasattr(content_item, 'text'): result += content_item.text else: result += str(item.content) result = _strip_markdown(result) # 【v2.0新增】保存群聊AI回复 if session_id and group_id and result: save_group_ai_reply( session_id=session_id, group_id=group_id, ai_role='femininity', content=result, response_id=response.id, trigger_message=request.message, trigger_user_id=request.user_id, timestamp=datetime.now(), 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="femininity", trigger_msg=request.message, ai_reply=result, app_name=request.app_name, )) if not result: save_chat_log( user_id=request.user_id, question=request.message, stream_mode=False, raw_response=repr(response), status="error", ) raise HTTPException(status_code=500, detail="AI未能生成回复") # 记录日志 save_chat_log( user_id=request.user_id, question=request.message, stream_mode=False, raw_response=repr(response), status="success", ) return {"data": result} except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Femininity chat error: {str(e)}") # AI社群角色对话 @router.post("/aiCommunity") async def ai_community_chat(request: RoleBasedChatRequest): """AI氛围社群角色对话接口""" try: if _should_block_role_message(request.message): raise HTTPException(status_code=500, detail="AI未能生成回复") ai_config = get_config_by_app_name(request.app_name) if not ai_config: raise HTTPException(status_code=404, detail=f"未找到appName '{request.app_name}' 的配置") client = get_client(request.app_name) knowledge_id = ai_config.get("knowledgeId") session_id = request.session_id group_id = request.group_id previous_response_id = None if session_id: previous_response_id = get_last_group_response_id(session_id, ai_role='ai_community') context_messages = [] if group_id: from app.db.tencent_im import im_client try: group_context = get_cached_group_context(group_id) if group_context is None: group_context = await im_client.get_group_context(group_id, max_messages=20) if group_context: cache_group_context(group_id, group_context, ttl=120) if group_context: for msg in group_context[-20:]: msg_user_id = msg['user_id'] content_parts = [] if msg.get('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', []): content_parts.append({"type": "input_image", "image_url": img_url}) for video in msg.get('videos', []): if video.get('thumb'): content_parts.append({"type": "input_image", "image_url": video['thumb']}) content_parts.append({"type": "input_text", "text": "[视频]"}) if msg.get('share_title'): content_parts.append({"type": "input_text", "text": f"[分享]{msg['share_title']}"}) if content_parts: context_messages.extend(content_parts) except Exception as e: print(f"获取群历史失败: {e}") # AI社群角色系统提示词 system_prompt = """ ## 角色设定 你是一名对人工智能领域有浓厚兴趣的群友，经常关注大模型、AI工具和AI行业动向。你在群里的身份是参与者，不是主持人或讲师。 对于你不掌握的具体数据、最新消息或未经核实的信息，明确说"不确定"或"我没看到最新的"，不编造。 ## 群聊场景 群成员以AI爱好者、开发者、创业者和科技从业者为主，日常讨论大模型进展、AI工具使用、马斯克动态、AI创业机会、提示词技巧等话题，偶尔也聊AI行业的争议和担忧。 ## 不回复的情况（最高优先级） 遇到以下内容保持沉默，不输出任何文字： 1. **自杀/自伤倾向** 2. **诈骗/违规推广**：承诺高收益、保本翻倍、拉人入群、传销话术 3. **极端内容**：煽动仇恨、极端政治、宗教争议、明显机器人测试 4. **与AI/科技完全无关的话题**（情感困惑、家庭纠纷、医疗问题等）：不强行回复，保持沉默 遇到以上内容直接保持沉默，不解释为什么不回复。 ## 回应原则 **技术/产品问题：** 给出你自己的判断，可以说"我觉得""我用下来感觉"，不要总用"业界认为"或"普遍观点"。 **行业动态讨论：** 有自己的看法，可以对某个方向表达支持或质疑，不是中立播报员。 **争议话题（AGI时间线、AI取代人、AI安全）：** 给出你认为站得住脚的角度，不回避，但也不下绝对结论。 **闲聊：** 不必强行回复，尤其是与AI话题完全无关时。 ## 互动风格 1. 只在被@、被提问、话题值得补充或群聊冷场时发言，不是每条消息都要回。 2. 说话像群友，不像科技媒体：用"这个模型"而不是"该模型"，用"感觉"而不是"据悉"。 3. 对马斯克、OpenAI、Anthropic、谷歌、字节、阿里等公司动态可以有态度，但不散布未经证实的消息。 4. 简单问题简短回答，不要分一二三四。 ## 专业边界 - AI技术问题给有条件的判断，不把你的理解说成权威结论。 - 涉及投资、法律、安全漏洞等敏感问题，建议咨询专业人士，不做保证性判断。 - 不声称自己是某位真实人物，不编造最新发布会或论文数据。 ## 输出格式 只输出可以直接发到群里的消息，不加解释、分析过程、角色标签或引号。 一次只发一条消息。 如果不需要回应，保持沉默（不输出任何内容）。 """ # 【v3.0】注入用户长期记忆 mem = get_user_memory(request.user_id, group_id, "ai_community") 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}]} user_content = [{"type": "input_text", "text": request.message}] if context_messages: context_header = [{"type": "input_text", "text": "【群聊上下文（最近20条消息）】"}] context_footer = [{"type": "input_text", "text": "\n\n当前用户发送的消息是："}] api_messages = [ system_msg, {"role": "user", "content": context_header + context_messages + context_footer + user_content} ] else: api_messages = [system_msg, {"role": "user", "content": user_content}] 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请优先基于以上资料回答，资料中没有的内容不要编造。"}] 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() response = client.responses.create( model="doubao-seed-2-1-lite-260915", input=api_messages, stream=False, store=True if session_id else False, previous_response_id=previous_response_id, text={"format": {"type": "text"}}, tools=tools, ) result = "" for item in response.output: if hasattr(item, 'type') and item.type == 'message' and hasattr(item, 'content'): if isinstance(item.content, list): for content_item in item.content: if hasattr(content_item, 'text'): result += content_item.text else: result += str(item.content) result = _strip_markdown(result) if session_id and group_id and result: save_group_ai_reply( session_id=session_id, group_id=group_id, ai_role='ai_community', content=result, response_id=response.id, trigger_message=request.message, trigger_user_id=request.user_id, timestamp=datetime.now(), app_name=request.app_name, ) if result: asyncio.create_task(_extract_memory( user_id=request.user_id, group_id=group_id, ai_role="ai_community", trigger_msg=request.message, ai_reply=result, app_name=request.app_name, )) if not result: save_chat_log( user_id=request.user_id, question=request.message, stream_mode=False, raw_response=repr(response), status="error", ) raise HTTPException(status_code=500, detail="AI未能生成回复") save_chat_log( user_id=request.user_id, question=request.message, stream_mode=False, raw_response=repr(response), status="success", ) return {"data": result} except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"AI Community chat error: {str(e)}") # 掼蛋角色对话 @router.post("/guandan") async def guandan_chat(request: RoleBasedChatRequest): """掼蛋社群角色对话接口""" try: if _should_block_role_message(request.message): raise HTTPException(status_code=500, detail="AI未能生成回复") ai_config = get_config_by_app_name(request.app_name) if not ai_config: raise HTTPException(status_code=404, detail=f"未找到appName '{request.app_name}' 的配置") client = get_client(request.app_name) knowledge_id = ai_config.get("knowledgeId") session_id = request.session_id group_id = request.group_id previous_response_id = None if session_id: previous_response_id = get_last_group_response_id(session_id, ai_role='guandan') context_messages = [] if group_id: from app.db.tencent_im import im_client try: group_context = get_cached_group_context(group_id) if group_context is None: group_context = await im_client.get_group_context(group_id, max_messages=20) if group_context: cache_group_context(group_id, group_context, ttl=120) if group_context: for msg in group_context[-20:]: msg_user_id = msg['user_id'] content_parts = [] if msg.get('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', []): content_parts.append({"type": "input_image", "image_url": img_url}) for video in msg.get('videos', []): if video.get('thumb'): content_parts.append({"type": "input_image", "image_url": video['thumb']}) content_parts.append({"type": "input_text", "text": "[视频]"}) if msg.get('share_title'): content_parts.append({"type": "input_text", "text": f"[分享]{msg['share_title']}"}) if content_parts: context_messages.extend(content_parts) except Exception as e: print(f"获取群历史失败: {e}") # 掼蛋角色系统提示词 system_prompt = """ ## 角色设定 你是一名掼蛋老手，对掼蛋规则、打法和赛事都很熟悉。你在群里的身份是参与者，不是裁判或教练。 对于具体赛事结果、职业选手信息或你不确定的内容，明确说"不太清楚"，不编造。 ## 群聊场景 群成员都是掼蛋爱好者，讨论掼蛋比赛动态、打牌技巧、常见牌型处理、组队配合、贡牌策略等话题，也会分享自己打牌的经历和感受。 ## 不回复的情况（最高优先级） 遇到以下内容保持沉默，不输出任何文字： 1. **自杀/自伤倾向** 2. **诈骗/违规推广**：承诺高收益、拉人入群、传销话术、赌博套利相关 3. **极端内容**：煽动仇恨、极端政治、宗教争议 4. **与掼蛋完全无关的话题**（情感困惑、工作问题、医疗健康等）：不强行回复 遇到以上内容直接保持沉默，不解释为什么不回复。 ## 回应原则 **技巧/打法问题：** 给出具体的操作建议，说"我遇到这情况一般会..."，不要泛泛而谈。 **赛事讨论：** 对比赛结果或精彩牌局可以有自己的看法，不是简单陈述。 **吐槽/经历分享：** 先共情，可以说"这局太难了"，再问问具体情况。 **规则疑问：** 直接说清楚，不绕弯子。 **闲聊：** 与掼蛋无关时可以不回。 ## 互动风格 1. 只在被@、被提问、话题值得补充或群聊冷场时发言。 2. 像群里的牌友一样说话，口语化，不像规则手册。 3. 可以用掼蛋常见口语，比如"打主""跑高""抓贡""炸弹""天王炸""进贡""还贡""贡牌""上贡"等。 4. 简单问题简短回答，不分条列点。 ## 专业边界 - 涉及金钱赌博、违规活动相关话题不参与，也不给建议。 - 不编造职业赛事结果或选手信息。 ## 输出格式 只输出可以直接发到群里的消息，不加解释、分析过程、角色标签或引号。 一次只发一条消息。 如果不需要回应，保持沉默（不输出任何内容）。 """ # 【v3.0】注入用户长期记忆 mem = get_user_memory(request.user_id, group_id, "guandan") 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}]} user_content = [{"type": "input_text", "text": request.message}] if context_messages: context_header = [{"type": "input_text", "text": "【群聊上下文（最近20条消息）】"}] context_footer = [{"type": "input_text", "text": "\n\n当前用户发送的消息是："}] api_messages = [ system_msg, {"role": "user", "content": context_header + context_messages + context_footer + user_content} ] else: api_messages = [system_msg, {"role": "user", "content": user_content}] 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请优先基于以上资料回答，资料中没有的内容不要编造。"}] 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() response = client.responses.create( model="doubao-seed-2-1-lite-260915", input=api_messages, stream=False, store=True if session_id else False, previous_response_id=previous_response_id, text={"format": {"type": "text"}}, tools=tools, ) result = "" for item in response.output: if hasattr(item, 'type') and item.type == 'message' and hasattr(item, 'content'): if isinstance(item.content, list): for content_item in item.content: if hasattr(content_item, 'text'): result += content_item.text else: result += str(item.content) result = _strip_markdown(result) if session_id and group_id and result: save_group_ai_reply( session_id=session_id, group_id=group_id, ai_role='guandan', content=result, response_id=response.id, trigger_message=request.message, trigger_user_id=request.user_id, timestamp=datetime.now(), app_name=request.app_name, ) if result: asyncio.create_task(_extract_memory( user_id=request.user_id, group_id=group_id, ai_role="guandan", trigger_msg=request.message, ai_reply=result, app_name=request.app_name, )) if not result: save_chat_log( user_id=request.user_id, question=request.message, stream_mode=False, raw_response=repr(response), status="error", ) raise HTTPException(status_code=500, detail="AI未能生成回复") save_chat_log( user_id=request.user_id, question=request.message, stream_mode=False, raw_response=repr(response), status="success", ) return {"data": result} except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Guandan chat error: {str(e)}")