diff --git a/backend/python/app/api/admin/achievement_admin.py b/backend/python/app/api/admin/achievement_admin.py index 2f24867..4cc2e58 100644 --- a/backend/python/app/api/admin/achievement_admin.py +++ b/backend/python/app/api/admin/achievement_admin.py @@ -284,7 +284,7 @@ async def claim_achievement_reward( # Mark reward as claimed _execute_raw(db, - "UPDATE user_achievements SET reward_claimed = 1, claimed_at = NOW() WHERE user_id = :uid AND achievement_id = :aid", + "UPDATE user_achievements SET reward_claimed = true, claimed_at = NOW() WHERE user_id = :uid AND achievement_id = :aid", {"uid": user_id, "aid": ach_id}) return {"code": 0, "message": "奖励领取成功"} diff --git a/backend/python/app/api/admin/analytics_admin.py b/backend/python/app/api/admin/analytics_admin.py index 2b68c4d..0f9f47a 100644 --- a/backend/python/app/api/admin/analytics_admin.py +++ b/backend/python/app/api/admin/analytics_admin.py @@ -316,7 +316,7 @@ async def create_dashboard( config_json = json.loads('[{"type":"stat","title":"暂无数据","metric":"none","chart_type":"number"}]') if is_default: - db.execute(text("UPDATE analytics_dashboards SET is_default=0")) + db.execute(text("UPDATE analytics_dashboards SET is_default=false")) db.execute(text(""" INSERT INTO analytics_dashboards (id, name, description, config, is_default, created_by) @@ -332,7 +332,7 @@ async def create_dashboard( @router.delete("/dashboards/{dashboard_id}") async def delete_dashboard(dashboard_id: str, db: Session = Depends(get_db), user=Depends(get_current_user)): """删除仪表盘""" - db.execute(text("DELETE FROM analytics_dashboards WHERE id=:id AND is_default=0"), {"id": dashboard_id}) + db.execute(text("DELETE FROM analytics_dashboards WHERE id=:id AND is_default=false"), {"id": dashboard_id}) db.commit() return {"code": 200, "data": {"message": "仪表盘已删除"}} diff --git a/backend/python/app/api/admin/billing_admin.py b/backend/python/app/api/admin/billing_admin.py index a3d84c0..3333fbf 100644 --- a/backend/python/app/api/admin/billing_admin.py +++ b/backend/python/app/api/admin/billing_admin.py @@ -445,7 +445,7 @@ async def billing_stats(db: Session = Depends(get_db), user=Depends(get_current_ SELECT DATE(created_at) as d, COALESCE(SUM(amount), 0) as rev FROM billing_transactions WHERE transaction_type='recharge' AND status='completed' - AND created_at >= DATE_SUB(NOW(), INTERVAL 7 DAY) + AND created_at >= NOW() - INTERVAL '7 DAY' GROUP BY DATE(created_at) ORDER BY d """)).fetchall() daily_revenue = [{"date": str(r[0]), "revenue": float(r[1])} for r in revenue_7d] diff --git a/backend/python/app/api/admin/code_exec_admin.py b/backend/python/app/api/admin/code_exec_admin.py index b98fd16..ae3b971 100644 --- a/backend/python/app/api/admin/code_exec_admin.py +++ b/backend/python/app/api/admin/code_exec_admin.py @@ -103,7 +103,7 @@ async def get_execution_stats(db: Session = Depends(get_db), _=Depends(get_curre try: total = _execute_raw(db, "SELECT COUNT(*) as cnt FROM code_exec_tasks")[0]["cnt"] today = _execute_raw(db, - "SELECT COUNT(*) as cnt FROM code_exec_tasks WHERE DATE(created_at) = CURDATE()")[0]["cnt"] + "SELECT COUNT(*) as cnt FROM code_exec_tasks WHERE DATE(created_at) = CURRENT_DATE")[0]["cnt"] success_count = _execute_raw(db, "SELECT COUNT(*) as cnt FROM code_exec_tasks WHERE exit_code = 0 AND error IS NULL")[0]["cnt"] failed_count = _execute_raw(db, diff --git a/backend/python/app/api/admin/distro_admin.py b/backend/python/app/api/admin/distro_admin.py index 51319d0..308cf89 100644 --- a/backend/python/app/api/admin/distro_admin.py +++ b/backend/python/app/api/admin/distro_admin.py @@ -703,7 +703,7 @@ echo "Installation complete!" {d.get('version', '1.0.0')} ## Installation -Run `bash install.sh` to install. +Run \"bash install.sh\" to install. ## License A valid license key is required. Contact your administrator for details. diff --git a/backend/python/app/api/admin/im_bot_admin.py b/backend/python/app/api/admin/im_bot_admin.py index eff21bf..a767a35 100644 --- a/backend/python/app/api/admin/im_bot_admin.py +++ b/backend/python/app/api/admin/im_bot_admin.py @@ -169,7 +169,7 @@ async def get_im_bot_stats(user=Depends(get_current_user)): today_stats = _row_to_dict(rows[0]) if rows else {"msg_in": 0, "msg_out": 0, "users": 0, "errors": 0} # Total routing rules - routes_count = _execute_raw(db, text("SELECT COUNT(*) as cnt FROM im_bot_routes WHERE is_active=1"))[0]._mapping['cnt'] + routes_count = _execute_raw(db, text("SELECT COUNT(*) as cnt FROM im_bot_routes WHERE is_active=true"))[0]._mapping['cnt'] return { "success": True, @@ -202,7 +202,7 @@ async def get_dashboard(user=Depends(get_current_user)): SUM(messages_out) as msg_out, SUM(unique_users) as users FROM im_bot_daily_stats - WHERE stat_date >= DATE_SUB(CURDATE(), INTERVAL 7 DAY) + WHERE stat_date >= CURRENT_DATE - INTERVAL '7 DAY' GROUP BY stat_date ORDER BY stat_date """)) @@ -217,7 +217,7 @@ async def get_dashboard(user=Depends(get_current_user)): LEFT JOIN ( SELECT channel_id, SUM(messages_in + messages_out) as msg_count FROM im_bot_daily_stats - WHERE stat_date >= DATE_SUB(CURDATE(), INTERVAL 30 DAY) + WHERE stat_date >= CURRENT_DATE - INTERVAL '30 DAY' GROUP BY channel_id ) s ON c.id = s.channel_id ORDER BY c.created_at DESC @@ -357,7 +357,7 @@ async def get_channel_detail(channel_id: str, user=Depends(get_current_user)): SELECT stat_date, messages_in, messages_out, unique_users, error_count FROM im_bot_daily_stats WHERE channel_id = :cid - AND stat_date >= DATE_SUB(CURDATE(), INTERVAL 7 DAY) + AND stat_date >= CURRENT_DATE - INTERVAL '7 DAY' ORDER BY stat_date """), {"cid": channel_id}) channel['stats_7d'] = _rows_to_dict(stats) @@ -428,7 +428,7 @@ async def activate_channel(channel_id: str, user=Depends(get_current_user)): db = SessionLocal() try: _execute_raw(db, text( - "UPDATE channels SET is_active=1, status='connected' WHERE id=:cid" + "UPDATE channels SET is_active=true, status='connected' WHERE id=:cid" ), {"cid": channel_id}) return {"success": True, "message": "Channel activated"} except Exception as e: @@ -443,7 +443,7 @@ async def deactivate_channel(channel_id: str, user=Depends(get_current_user)): db = SessionLocal() try: _execute_raw(db, text( - "UPDATE channels SET is_active=0, status='disconnected' WHERE id=:cid" + "UPDATE channels SET is_active=false, status='disconnected' WHERE id=:cid" ), {"cid": channel_id}) return {"success": True, "message": "Channel deactivated"} except Exception as e: diff --git a/backend/python/app/api/admin/privacy_admin.py b/backend/python/app/api/admin/privacy_admin.py index 522a704..7156f88 100644 --- a/backend/python/app/api/admin/privacy_admin.py +++ b/backend/python/app/api/admin/privacy_admin.py @@ -442,7 +442,7 @@ def revoke_license( raise HTTPException(404, "License not found") _execute_raw(db, text( - "UPDATE offline_licenses SET status = 'revoked', revoked_reason = :reason, is_active = 0 WHERE id = :lid" + "UPDATE offline_licenses SET status = 'revoked', revoked_reason = :reason, is_active = false WHERE id = :lid" ), {"reason": reason, "lid": license_id}) return {"code": 0, "message": "License revoked"} @@ -623,7 +623,7 @@ def get_usage_stats(db: Session = Depends(get_db)): "SELECT COUNT(DISTINCT user_hash) FROM usage_reports" )) today_reports = _execute_raw(db, text( - "SELECT COUNT(*) FROM usage_reports WHERE DATE(created_at) = CURDATE()" + "SELECT COUNT(*) FROM usage_reports WHERE DATE(created_at) = CURRENT_DATE" )) by_type = _execute_raw(db, text( "SELECT report_type, COUNT(*) as cnt FROM usage_reports GROUP BY report_type" diff --git a/backend/python/app/api/chat_context.py b/backend/python/app/api/chat_context.py index 29e0008..3289d03 100644 --- a/backend/python/app/api/chat_context.py +++ b/backend/python/app/api/chat_context.py @@ -1251,7 +1251,7 @@ def _build_system_prompt( - system_parts.append("回复格式要求：使用 Markdown 格式，层次分明即可。对于简短问题，用 1-2 段自然文字直接回答，不必强行分点、加标题或表格；仅在确实需要分步骤或列举时才使用列表，代码/配置用 ``` 代码块包裹。避免冗余空行，保持紧凑、美观、易读。") + system_parts.append("回复格式要求：使用 Markdown 格式，层次分明即可。对于简短问题，用 1-2 段自然文字直接回答，不必强行分点、加标题或表格；仅在确实需要分步骤或列举时才使用列表，代码/配置用 \"\"\" 代码块包裹。避免冗余空行，保持紧凑、美观、易读。") @@ -2536,7 +2536,6 @@ def _build_chat_history(db, conv_id, max_tokens=6000, max_messages=40, - 按 token 估算从新到旧累加，超预算即截断（隔离：不溢出上下文窗口）。 - exclude_latest_user=True 时丢弃最新一条 user 消息（当前轮次另行发送）。 """ - ensure_structured_column() summary_block = None if include_summary: try: @@ -2707,7 +2706,7 @@ def ensure_structured_column(): with engine.connect() as conn: exists = conn.execute(text( "SELECT COUNT(*) FROM information_schema.columns " - "WHERE table_schema = DATABASE() AND table_name='conversation_summaries' " + "WHERE table_schema = CURRENT_DATABASE() AND table_name='conversation_summaries' " "AND column_name='structured_memory'" )).scalar() if not exists: @@ -2832,7 +2831,6 @@ async def _maybe_compress_history(conv_id): 失败静默回滚，绝不阻塞用户本次响应（fire-and-forget）。 """ - ensure_structured_column() sdb = None try: sdb = SessionLocal() diff --git a/backend/python/app/api/chat_media.py b/backend/python/app/api/chat_media.py index ed40b18..29537dd 100644 --- a/backend/python/app/api/chat_media.py +++ b/backend/python/app/api/chat_media.py @@ -3,7 +3,8 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Response from sqlalchemy.orm import Session -from sqlalchemy import and_, desc, asc, func +from sqlalchemy import and_, desc, asc, func, type_coerce +from sqlalchemy.dialects.postgresql import JSONB from pydantic import BaseModel, Field from typing import Optional import uuid @@ -893,7 +894,7 @@ def _build_media_route(db, capability: str, preferred_id: str = None): others = db.query(_AM).filter( _AM.is_enabled == True, - func.json_contains(_AM.capabilities, f'"{capability}"'), + _AM.capabilities.cast(JSONB).contains(type_coerce([capability], JSONB)), ).order_by(_AM.is_default.desc(), _AM.created_at.asc()).all() for m in others: _add(m) diff --git a/backend/python/app/api/chat_service.py b/backend/python/app/api/chat_service.py index ccc1108..7fd230b 100644 --- a/backend/python/app/api/chat_service.py +++ b/backend/python/app/api/chat_service.py @@ -47,6 +47,7 @@ import json import asyncio +import threading @@ -7927,7 +7928,7 @@ async def _send_message_core( - asyncio.create_task( + _launch_media_task_bg(ai_msg.id, ai_model.id, req.content, current_user.id) @@ -7943,7 +7944,6 @@ async def _send_message_core( - _generate_media_task(ai_msg.id, ai_model.id, req.content, current_user.id) @@ -7959,7 +7959,6 @@ async def _send_message_core( - ) @@ -8547,3 +8546,16 @@ def _check_tool_policy(func_name, mcp_info, skills, policy, call_count, round_co + + +# 后台线程独立事件循环运行媒体生成，确保 SSE 响应结束后任务不被 uviorn 取消， +# 避免图片/视频永远停留在「生成中…」。_generate_media_task 内部自管 DB 会话。 +def _launch_media_task_bg(ai_msg_id, ai_model_id, prompt, user_id): + def _run(): + try: + asyncio.run(_generate_media_task(ai_msg_id, ai_model_id, prompt, user_id)) + except Exception: + pass + _t = threading.Thread(target=_run, daemon=True) + _t.start() + diff --git a/backend/python/app/api/im_bot_platform.py b/backend/python/app/api/im_bot_platform.py index e8a0d4a..3fb0170 100644 --- a/backend/python/app/api/im_bot_platform.py +++ b/backend/python/app/api/im_bot_platform.py @@ -109,20 +109,20 @@ def _ensure_table(db): """确保 im_bot_platform_account 表存在（首次调用时创建）。""" db.execute(text(""" CREATE TABLE IF NOT EXISTS im_bot_platform_account ( - id INT AUTO_INCREMENT PRIMARY KEY, + id SERIAL PRIMARY KEY, user_id VARCHAR(64) DEFAULT NULL, platform VARCHAR(32) NOT NULL, status VARCHAR(32) DEFAULT 'disconnected', - creds MEDIUMTEXT NULL, + creds TEXT NULL, send_url VARCHAR(512) NULL, api_base VARCHAR(512) NULL, bot_token VARCHAR(512) NULL, webhook_token VARCHAR(64) NULL, - last_error MEDIUMTEXT NULL, - created_at DATETIME DEFAULT CURRENT_TIMESTAMP, - updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, - UNIQUE KEY uk_user_platform (user_id, platform) - ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci + last_error TEXT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + UNIQUE (user_id, platform) + ) """)) db.commit() diff --git a/backend/python/app/api/knowledge.py b/backend/python/app/api/knowledge.py index d708351..5315ed4 100644 --- a/backend/python/app/api/knowledge.py +++ b/backend/python/app/api/knowledge.py @@ -502,8 +502,8 @@ def _parse_llm_graph_response(content: str): except json.JSONDecodeError: pass - # 方式2: 提取```json ... ```代码块 - json_match = re.search(r'```(?:json)?\s*([\s\S]*?)```', content) + # 方式2: 提取\"\"\"json ... \"\"\"代码块 + json_match = re.search(r'\"\"\"(?:json)?\s*([\s\S]*?)\"\"\"', content) if json_match: try: result = json.loads(json_match.group(1)) diff --git a/backend/python/app/api/market.py b/backend/python/app/api/market.py index 9e0e42b..933eaec 100644 --- a/backend/python/app/api/market.py +++ b/backend/python/app/api/market.py @@ -59,7 +59,7 @@ def _sanitize_skill_content(text: str) -> str: if " 3: return "" # 接受携带任一 markdown / skill 标记的内容 - markers = ("#", "---", "name:", "description:", "```", "##") + markers = ("#", "---", "name:", "description:", "\"\"\"", "##") if not any(m in t for m in markers): return "" return t @@ -79,7 +79,7 @@ def _extract_skill_from_html(html: str, base_url: str) -> str: """从 HTML 页面尽力抽取 SKILL.md 正文（代码块优先，其次跳转到 raw 链接）。""" proxy = _skill_http_proxy() # 1) fenced code block - fences = re.findall(r"```(?:markdown|md|yaml|json)?\s*\n(.*?)```", html, re.DOTALL | re.IGNORECASE) + fences = re.findall(r"\"\"\"(?:markdown|md|yaml|json)?\s*\n(.*?)\"\"\"", html, re.DOTALL | re.IGNORECASE) for f in fences: if ("name:" in f.lower() or "description:" in f.lower() or f.count("#") >= 2) and len(f) > 80: return f.strip() diff --git a/backend/python/app/capability_registry.py b/backend/python/app/capability_registry.py index 8ac735e..ea93980 100644 --- a/backend/python/app/capability_registry.py +++ b/backend/python/app/capability_registry.py @@ -98,10 +98,11 @@ def resolve_auto_model(capability: str, db: Session) -> Optional[AIModel]: # 媒体类能力(image/video/audio)：无显式配置时自动发现任一启用的模型 # 避免用户未在后台配置"能力默认模型"时 Auto 回退到文本模型导致"无法生成图片/视频" if capability in ("image", "video", "audio"): - from sqlalchemy import text as _text, func + from sqlalchemy import text as _text, func, type_coerce + from sqlalchemy.dialects.postgresql import JSONB m = db.query(AIModel).filter( AIModel.is_enabled == True, - func.json_contains(AIModel.capabilities, f'"{capability}"'), + AIModel.capabilities.cast(JSONB).contains(type_coerce([capability], JSONB)), ).first() if m: return m diff --git a/backend/python/app/chat_context.py b/backend/python/app/chat_context.py index bb4173b..d45d58d 100644 --- a/backend/python/app/chat_context.py +++ b/backend/python/app/chat_context.py @@ -1293,7 +1293,7 @@ def _build_system_prompt( - system_parts.append("回复格式要求：使用 Markdown 格式，层次分明即可。对于简短问题，用 1-2 段自然文字直接回答，不必强行分点、加标题或表格；仅在确实需要分步骤或列举时才使用列表，代码/配置用 ``` 代码块包裹。避免冗余空行，保持紧凑、美观、易读。") + system_parts.append("回复格式要求：使用 Markdown 格式，层次分明即可。对于简短问题，用 1-2 段自然文字直接回答，不必强行分点、加标题或表格；仅在确实需要分步骤或列举时才使用列表，代码/配置用 \"\"\" 代码块包裹。避免冗余空行，保持紧凑、美观、易读。") @@ -2749,7 +2749,7 @@ def ensure_structured_column(): with engine.connect() as conn: exists = conn.execute(text( "SELECT COUNT(*) FROM information_schema.columns " - "WHERE table_schema = DATABASE() AND table_name='conversation_summaries' " + "WHERE table_schema = CURRENT_DATABASE() AND table_name='conversation_summaries' " "AND column_name='structured_memory'" )).scalar() if not exists: diff --git a/backend/python/app/config.py b/backend/python/app/config.py index 2582c50..e13f8b3 100644 --- a/backend/python/app/config.py +++ b/backend/python/app/config.py @@ -15,7 +15,12 @@ if not (Path("/.dockerenv").exists() or Path("/run/.containerenv").exists()): load_dotenv(env_path, override=False) # 数据库配置 -DATABASE_URL = f"mysql+pymysql://{os.getenv('MYSQL_USER', 'root')}:{os.getenv('MYSQL_PASSWORD', 'pwd2026')}@{os.getenv('MYSQL_HOST', 'localhost')}:{os.getenv('MYSQL_PORT', '3306')}/{os.getenv('MYSQL_DATABASE', 'aiflow')}?charset=utf8mb4" +# PostgreSQL preferred when DATABASE_URL is provided (production); else legacy MySQL builder (local/dev + rollback) +_DATABASE_URL_ENV = os.getenv("DATABASE_URL", "") +if _DATABASE_URL_ENV: + DATABASE_URL = _DATABASE_URL_ENV +else: + DATABASE_URL = f"mysql+pymysql://{os.getenv('MYSQL_USER', 'root')}:{os.getenv('MYSQL_PASSWORD', 'pwd2026')}@{os.getenv('MYSQL_HOST', 'localhost')}:{os.getenv('MYSQL_PORT', '3306')}/{os.getenv('MYSQL_DATABASE', 'aiflow')}?charset=utf8mb4" # Redis配置 REDIS_HOST = os.getenv("REDIS_HOST", "localhost") diff --git a/backend/python/app/core/llm_client.py b/backend/python/app/core/llm_client.py index ff2cfaf..cc2e217 100644 --- a/backend/python/app/core/llm_client.py +++ b/backend/python/app/core/llm_client.py @@ -1,9 +1,9 @@ """Shared, standards-based LLM client for web chat + IM bot. 平台红线(07-24)：模型层以成熟/开源标准为设计基准。本模块使用官方 OpenAI SDK -(`openai.AsyncOpenAI`)——它是 OpenAI 兼容端点的事实标准客户端。平台本身已在讲 -OpenAI 协议(`/v1/chat/completions`)，因此零协议改造成本，并天然获得： - * 原生 SSE 流式(`stream=True`)； +(\"openai.AsyncOpenAI\")——它是 OpenAI 兼容端点的事实标准客户端。平台本身已在讲 +OpenAI 协议(\"/v1/chat/completions\")，因此零协议改造成本，并天然获得： + * 原生 SSE 流式(\"stream=True\")； * 结构化的 tool_calls / usage 解析； * 统一的代理(trust_env=False + 显式 proxy)与超时处理。 @@ -13,13 +13,14 @@ IM bot 共用**同一模型层**、同一套 failover/代理/超时语义—— API: achat_complete(api_url, api_key, request_body, *, proxy=None, timeout=120, stream=False) - - stream=False -> dict，形状对齐上游 `data` 响应： + - stream=False -> dict，形状对齐上游 \"data\" 响应： {"choices":[{"message":{"content","tool_calls","reasoning_content"}}], "usage":{...}} - stream=True -> async generator，逐块产出 ("content"|"reasoning", text) 增量 """ from __future__ import annotations import httpx +import asyncio from openai import AsyncOpenAI, OpenAI @@ -37,10 +38,21 @@ def build_openai_client(api_url, api_key, *, proxy=None, timeout=120) -> AsyncOp # 始终 trust_env=False：避免容器 HTTP_PROXY 对国内端点(siliconflow 等)造成 502； # 代理经 proxy 显式传入（AGNES 等），未传则直连。 http_client = httpx.AsyncClient(proxy=proxy, trust_env=False) + # 故障快速失败（全栈统一，覆盖 web chat / IM bot / embedding）： + # connect 封顶 10s —— 坏端点不再让请求挂死事件循环； + # read/write 封顶 30s —— 长流式回复只要持续有分块即不受影响； + # 保留数值型 timeout 接口，旧调用方无需改动。 + _num = float(timeout) if isinstance(timeout, (int, float)) else 120.0 + _to = httpx.Timeout( + connect=min(10.0, _num), + read=min(30.0, _num), + write=min(30.0, _num), + pool=10.0, + ) return AsyncOpenAI( base_url=base, api_key=api_key or "EMPTY", - timeout=timeout, + timeout=_to, http_client=http_client, ) @@ -94,10 +106,21 @@ async def achat_complete(api_url, api_key, request_body, *, proxy=None, timeout= _reasoning_parts = [] _tool_acc = {} # index -> {id, type, name, arguments} try: - gen = await client.chat.completions.create( - **_create_kwargs(request_body, stream=True) + _to = min(30.0, float(timeout) if isinstance(timeout, (int, float)) else 120.0) + gen = await asyncio.wait_for( + client.chat.completions.create( + **_create_kwargs(request_body, stream=True) + ), + timeout=_to, ) - async for chunk in gen: + _it = gen.__aiter__() + while True: + try: + chunk = await asyncio.wait_for(_it.__anext__(), timeout=_to) + except StopAsyncIteration: + break + except asyncio.TimeoutError: + raise TimeoutError("stream chunk read timeout") if not chunk.choices: continue delta = chunk.choices[0].delta @@ -129,7 +152,10 @@ async def achat_complete(api_url, api_key, request_body, *, proxy=None, timeout= if _fn_args: _slot["arguments"] += _fn_args finally: - await client.close() + try: + await asyncio.wait_for(client.close(), timeout=5) + except Exception: + pass # 流式结束时吐出已组装的完整 message（含 tool_calls），供调用方无缝衔接工具循环 _tool_calls = None if _tool_acc: @@ -150,8 +176,12 @@ async def achat_complete(api_url, api_key, request_body, *, proxy=None, timeout= return _gen() client = build_openai_client(api_url, api_key, proxy=proxy, timeout=timeout) - async with client: - resp = await client.chat.completions.create(**_create_kwargs(request_body)) + try: + _to = min(30.0, float(timeout) if isinstance(timeout, (int, float)) else 120.0) + resp = await asyncio.wait_for( + client.chat.completions.create(**_create_kwargs(request_body)), + timeout=_to, + ) msg = resp.choices[0].message return { "choices": [ @@ -165,6 +195,11 @@ async def achat_complete(api_url, api_key, request_body, *, proxy=None, timeout= ], "usage": (resp.usage.model_dump() if resp.usage else {}), } + finally: + try: + await asyncio.wait_for(client.close(), timeout=5) + except Exception: + pass def build_openai_client_sync(api_url, api_key, *, proxy=None, timeout=120) -> OpenAI: @@ -208,9 +243,18 @@ async def aembed(api_url, api_key, input, model, *, proxy=None, timeout=60) -> l 嵌入调用收敛到同一标准模型层（与聊天共用代理/超时/failover 语义）。 """ client = build_openai_client(api_url, api_key, proxy=proxy, timeout=timeout) - async with client: - resp = await client.embeddings.create(**_embed_kwargs(input, model)) + try: + _to = min(60.0, float(timeout) if isinstance(timeout, (int, float)) else 60.0) + resp = await asyncio.wait_for( + client.embeddings.create(**_embed_kwargs(input, model)), + timeout=_to, + ) return [d.embedding for d in resp.data] + finally: + try: + await asyncio.wait_for(client.close(), timeout=5) + except Exception: + pass def embed(api_url, api_key, input, model, *, proxy=None, timeout=60) -> list: diff --git a/backend/python/app/core/observability.py b/backend/python/app/core/observability.py index a1521d4..fb7af49 100644 --- a/backend/python/app/core/observability.py +++ b/backend/python/app/core/observability.py @@ -6,7 +6,7 @@ Qihu-AIFlow 统一可观测基座 (Phase 2.1 + Phase 1.I OTel 增强) 设计原则（Strangler Fig 增量）： - 纯增量埋点，不改任何既有接口的返回结构。 -- 向后兼容现有 `logging.getLogger(__name__)` 用法：trace_id 由 root logger 的 +- 向后兼容现有 \"logging.getLogger(__name__)\" 用法：trace_id 由 root logger 的 TraceIdFilter 自动注入，调用方无需改动。 - 不引入第三方依赖（structlog 未安装），自带可选的 JSON formatter。 @@ -58,7 +58,7 @@ def get_trace_id() -> str: class TraceIdFilter(logging.Filter): """为每一条日志记录注入 trace_id 属性（不存在则写为 '-'）。 - 挂载到 root logger 后，所有 `getLogger(__name__)` 的子 logger 都会继承， + 挂载到 root logger 后，所有 \"getLogger(__name__)\" 的子 logger 都会继承， 从而让全平台既有日志自动带上链路 ID，零侵入。 """ diff --git a/backend/python/app/mcp_client.py b/backend/python/app/mcp_client.py index 0bc85be..8a4d24f 100644 --- a/backend/python/app/mcp_client.py +++ b/backend/python/app/mcp_client.py @@ -1,8 +1,8 @@ """ -MCP (Model Context Protocol) 客户端 —— 基于官方 `mcp` Python SDK (Phase 1.E 标准化) +MCP (Model Context Protocol) 客户端 —— 基于官方 \"mcp\" Python SDK (Phase 1.E 标准化) 设计基准（标准对照表见 STANDARDS_AUDIT_2026-07-24.md §E）： -- 采用 Anthropic 官方 `mcp` SDK 处理 JSON-RPC 帧 + 协议版本协商， +- 采用 Anthropic 官方 \"mcp\" SDK 处理 JSON-RPC 帧 + 协议版本协商， 不再手写解析、不再写死 MCP_PROTOCOL_VERSION 常量（旧实现写死 "2024-11-05"， 上游协议一升级即碎）。 - 支持 streamable_http（默认）与 sse 两种传输，按 server['transport_type'] 选择。 diff --git a/backend/python/app/models.py b/backend/python/app/models.py index efc8a29..f1bef60 100644 --- a/backend/python/app/models.py +++ b/backend/python/app/models.py @@ -11,7 +11,7 @@ from sqlalchemy import ( DECIMAL, UniqueConstraint ) from sqlalchemy.orm import relationship -from sqlalchemy.dialects.mysql import MEDIUMTEXT +from sqlalchemy import Text # PG-compatible (was MEDIUMTEXT from mysql dialect) from .database import Base @@ -300,7 +300,7 @@ class MarketApp(Base): rating_avg = Column(DECIMAL(3, 2), default=0.00) review_count = Column(Integer, default=0) install_count = Column(Integer, default=0) - skill_content = Column(MEDIUMTEXT, default=None) + skill_content = Column(Text, default=None) mcp_transport = Column(String(20), default=None) platform = Column(String(20), default=None, comment="Agent 接入平台：dify/n8n/coze") capability = Column(String(20), default=None, comment="能力分类：text/thinking/vision/tool/image/audio/asr/ocr/embedding/rerank/video") diff --git a/backend/python/app/platform_knowledge.py b/backend/python/app/platform_knowledge.py index 25df381..a20c720 100644 --- a/backend/python/app/platform_knowledge.py +++ b/backend/python/app/platform_knowledge.py @@ -73,7 +73,7 @@ PLATFORM_KNOWLEDGE = """# 七虎AI平台能力全览 - **安全**: JWT鉴权、HTTPS加密、操作审计日志 ## AI回复规范 -1. **回复格式**: 使用Markdown（标题 `##`/`###`、加粗 `**`、列表 `-`/`1.`、代码块等） +1. **回复格式**: 使用Markdown（标题 \"##\"/\"###\"、加粗 \"**\"、列表 \"-\"/\"1.\"、代码块等） 2. **层次分明**: 使用标题分隔段落，确保信息有层次 3. **平台优先**: 回答问题时优先引用本平台功能，而非泛泛建议 4. **准确具体**: 基于平台文档给出具体操作路径、配置方法、API示例 diff --git a/backend/python/main.py b/backend/python/main.py index 890087e..4c8950e 100644 --- a/backend/python/main.py +++ b/backend/python/main.py @@ -118,6 +118,18 @@ def _record_api_call(db, method: str, path: str, status_code: int, elapsed: floa async def lifespan(app: FastAPI): logger.info("Qihu-AIFlow v2.0.0 启动中...") + # 幂等确保 conversation_summaries.structured_memory 列。 + # 原实现在每次聊天的请求路径中同步执行 engine.connect().execute(SELECT information_schema.columns)， + # 该同步阻塞调用运行在 asyncio 事件循环线程上，会导致整个事件循环冻结（所有端点挂死， + # 表现为「单 worker 永久阻塞、docker restart 才恢复」）。改为启动时一次性执行， + # 既保留对全新库的兼容，又绝不阻塞任何用户请求。 + try: + from app.api.chat_context import ensure_structured_column + ensure_structured_column() + logger.info("[startup] ensure_structured_column 完成") + except Exception as _e: + logger.warning(f"[startup] ensure_structured_column 失败(可忽略): {_e}") + # 启动后台 Evolution Pipeline（知识库图谱已存在的用户配置自动恢复运行） try: from app.services.evolution_service import EvolutionWatcherManager diff --git a/backend/python/requirements.txt b/backend/python/requirements.txt index 1f19771..c87a156 100644 --- a/backend/python/requirements.txt +++ b/backend/python/requirements.txt @@ -28,3 +28,4 @@ click==8.4.2 click-didyoumean==0.3.1 click-plugins==1.1.1.2 click-repl==0.3.0 +psycopg[binary] diff --git a/backend/python/workflow_templates/code-review.yaml b/backend/python/workflow_templates/code-review.yaml new file mode 100644 index 0000000..b077d39 --- /dev/null +++ b/backend/python/workflow_templates/code-review.yaml @@ -0,0 +1,35 @@ +version: "1.0" +name: "Code Review Pipeline" +description: "Generate code → review → merge. A standard code-review workflow." +category: development + +nodes: + - id: code_gen + type: llm + description: "Generate code based on requirements" + on_error: halt + config: + model: auto + system_prompt: "You are a senior software engineer. Write clean, production-ready code based on the user's requirements. Include comments and error handling." + + - id: review + type: llm + description: "Review generated code for bugs and style" + on_error: continue + config: + model: auto + system_prompt: "You are a code reviewer. Analyze the provided code for bugs, style issues, performance problems, and security vulnerabilities. Provide specific, actionable feedback." + + - id: merge + type: reduce + description: "Combine code with review feedback" + config: + strategy: concat + +edges: + - from: code_gen + to: review + - from: code_gen + to: merge + - from: review + to: merge diff --git a/backend/python/workflow_templates/multi-agent-debate.yaml b/backend/python/workflow_templates/multi-agent-debate.yaml new file mode 100644 index 0000000..07bb7bf --- /dev/null +++ b/backend/python/workflow_templates/multi-agent-debate.yaml @@ -0,0 +1,35 @@ +version: "1.0" +name: "Multi-Agent Debate" +description: "Two agents debate a topic from opposing perspectives, then a judge agent synthesizes the conclusion." +category: creativity + +nodes: + - id: proponent + type: llm + description: "Argue in favor of the proposition" + on_error: continue + config: + model: auto + system_prompt: "You are a skilled debater. Argue strongly IN FAVOR of the given proposition. Use logic, evidence, and persuasive reasoning. Be passionate but fair." + + - id: opponent + type: llm + description: "Argue against the proposition" + on_error: continue + config: + model: auto + system_prompt: "You are a skilled debater. Argue strongly AGAINST the given proposition. Use logic, evidence, and persuasive reasoning to find flaws in the proponent's case." + + - id: judge + type: llm + description: "Evaluate both sides and reach a conclusion" + on_error: continue + config: + model: auto + system_prompt: "You are an impartial judge. Evaluate both arguments fairly. Identify the strongest points from each side. Provide a balanced conclusion and recommendation." + +edges: + - from: proponent + to: judge + - from: opponent + to: judge diff --git a/backend/python/workflow_templates/rag-search.yaml b/backend/python/workflow_templates/rag-search.yaml new file mode 100644 index 0000000..db8c297 --- /dev/null +++ b/backend/python/workflow_templates/rag-search.yaml @@ -0,0 +1,25 @@ +version: "1.0" +name: "RAG Search Pipeline" +description: "Search knowledge base → augment with context → generate answer. Standard RAG pipeline." +category: productivity + +nodes: + - id: search + type: llm + description: "Generate search queries and retrieve relevant context" + on_error: continue + config: + model: auto + system_prompt: "You are a search specialist. For the given question, identify 2-3 key search points. Then search the provided knowledge base for relevant information. Return the most relevant passages with source references." + + - id: answer + type: llm + description: "Generate final answer with citations" + on_error: continue + config: + model: auto + system_prompt: "You are a knowledge assistant. Using the search results provided, generate a comprehensive answer. Always cite your sources. If information is insufficient, clearly state what is missing." + +edges: + - from: search + to: answer diff --git a/backend/python/workflow_templates/research-assistant.yaml b/backend/python/workflow_templates/research-assistant.yaml new file mode 100644 index 0000000..73b9b54 --- /dev/null +++ b/backend/python/workflow_templates/research-assistant.yaml @@ -0,0 +1,25 @@ +version: "1.0" +name: "Research & Answer" +description: "Research topic → generate answer. Uses two LLM nodes for research and synthesis." +category: productivity + +nodes: + - id: research + type: llm + description: "Search and gather information on the topic" + on_error: continue + config: + model: auto + system_prompt: "You are a thorough researcher. Search for key facts, data points, and relevant context about the given topic. Provide a comprehensive summary with citations." + + - id: answer + type: llm + description: "Synthesize research into final answer" + on_error: continue + config: + model: auto + system_prompt: "You are a professional writer. Based on the research summary provided, craft a clear, well-structured answer for the user. Include key findings and actionable insights." + +edges: + - from: research + to: answer