Delete service_chat.py

service_chat.py

@@ -1,343 +0,0 @@
-import json
-import asyncio
-import re
-import tiktoken
-from config import BAI_API_KEY, BAI_URL, MODEL_NAME
-from tools_def import get_filtered_tools
-from service_connectors import execute_tool_call, execute_tavily_search
-
-enc = tiktoken.get_encoding("cl100k_base")
-
-def count_tokens(messages):
-    num_tokens = 0
-    for message in messages:
-        num_tokens += 4
-        for key, value in message.items():
-            if isinstance(value, str):
-                num_tokens += len(enc.encode(value))
-            elif isinstance(value, list):
-                for item in value:
-                    if isinstance(item, dict):
-                        if item.get("type") == "text":
-                            num_tokens += len(enc.encode(item.get("text", "")))
-                        elif item.get("type") == "image_url":
-                            num_tokens += 200
-    num_tokens += 2
-    return num_tokens
-
-def normalize_query_for_cache(q: str) -> str:
-    cleaned = re.sub(r"[^\w\s]", "", q.lower()).strip()
-    return re.sub(r"\s+", " ", cleaned)
-
-def format_image_url(img_str):
-    if not isinstance(img_str, str) or not img_str.strip():
-        return None
-    img_clean = img_str.strip()
-    if img_clean.startswith("http://") or img_clean.startswith("https://") or img_clean.startswith("data:"):
-        return img_clean
-    return f"data:image/jpeg;base64,{img_clean}"
-
-def normalize_content(content):
-    if isinstance(content, str):
-        return content.strip()
-    if isinstance(content, list):
-        normalized = []
-        for item in content:
-            if isinstance(item, dict):
-                t = item.get("type")
-                if t == "text" and item.get("text"):
-                    normalized.append({"type": "text", "text": str(item.get("text", "")).strip()})
-                elif t == "image_url":
-                    url = item.get("image_url", {}).get("url", "")
-                    fmt_url = format_image_url(url)
-                    if fmt_url:
-                        normalized.append({"type": "image_url", "image_url": {"url": fmt_url}})
-                elif "image" in item:
-                    fmt_url = format_image_url(item.get("image", ""))
-                    if fmt_url:
-                        normalized.append({"type": "image_url", "image_url": {"url": fmt_url}})
-        return normalized
-    return ""
-
-def build_context_messages(req_messages, user_prompt, images=None, max_tokens=100000):
-    raw_list = []
-    
-    if req_messages and isinstance(req_messages, list) and len(req_messages) > 0:
-        raw_list = req_messages
-    elif user_prompt or images:
-        content_parts = []
-        if user_prompt:
-            content_parts.append({"type": "text", "text": user_prompt})
-        if images and isinstance(images, list):
-            for img in images:
-                fmt_url = format_image_url(img)
-                if fmt_url:
-                    content_parts.append({
-                        "type": "image_url",
-                        "image_url": {"url": fmt_url}
-                    })
-        if len(content_parts) == 1 and content_parts[0]["type"] == "text":
-            raw_list = [{"role": "user", "content": content_parts[0]["text"]}]
-        else:
-            raw_list = [{"role": "user", "content": content_parts}]
-
-    cleaned = []
-    for msg in raw_list:
-        if not isinstance(msg, dict):
-            continue
-        role = msg.get("role")
-        content = normalize_content(msg.get("content"))
-        if not role or not content:
-            continue
-        cleaned.append({"role": role, "content": content})
-
-    if images and isinstance(images, list) and len(images) > 0 and cleaned:
-        last_msg = cleaned[-1]
-        if last_msg.get("role") == "user":
-            current_content = last_msg.get("content")
-            if isinstance(current_content, str):
-                content_parts = [{"type": "text", "text": current_content}] if current_content else []
-            elif isinstance(current_content, list):
-                content_parts = list(current_content)
-            else:
-                content_parts = []
-            
-            for img in images:
-                fmt_url = format_image_url(img)
-                if fmt_url and not any(p.get("type") == "image_url" and p.get("image_url", {}).get("url") == fmt_url for p in content_parts):
-                    content_parts.append({
-                        "type": "image_url",
-                        "image_url": {"url": fmt_url}
-                    })
-            last_msg["content"] = content_parts
-
-    while cleaned and count_tokens(cleaned) > max_tokens:
-        cleaned.pop(0)
-        while cleaned and cleaned[0]["role"] != "user":
-            cleaned.pop(0)
-
-    return cleaned
-
-def get_tool_metadata(func_name: str, args: dict):
-    if func_name == "tavily_search":
-        q = args.get("query", "the web")
-        depth = args.get("search_depth", "basic")
-        if depth == "advanced":
-            return "search", f"Deep Research: {q}"
-        return "search", f"Searching: {q}"
-
-    endpoint = args.get("endpoint", "")
-    method = args.get("method", "GET")
-
-    if func_name == "gmail_api_request":
-        if "messages" in endpoint:
-            return "gmail", f"{method} Gmail messages"
-        elif "drafts" in endpoint:
-            return "gmail", f"{method} Gmail draft"
-        elif "labels" in endpoint:
-            return "gmail", f"{method} Gmail label"
-        return "gmail", f"{method} Gmail API"
-
-    elif func_name == "calendar_api_request":
-        if "events" in endpoint:
-            return "calendar", f"{method} Calendar events"
-        return "calendar", f"{method} Google Calendar API"
-
-    elif func_name == "drive_api_request":
-        if "files" in endpoint:
-            return "drive", f"{method} Drive files"
-        return "drive", f"{method} Google Drive API"
-
-    elif func_name == "notion_api_request":
-        if "search" in endpoint:
-            return "notion", "Search Notion workspace"
-        elif "pages" in endpoint:
-            return "notion", f"{method} Notion page"
-        elif "databases" in endpoint:
-            return "notion", f"{method} Notion database"
-        elif "blocks" in endpoint:
-            return "notion", f"{method} Notion blocks"
-        return "notion", f"{method} Notion API"
-
-    elif func_name == "github_api_request":
-        if "issues" in endpoint:
-            return "github", f"{method} GitHub issue"
-        elif "pulls" in endpoint:
-            return "github", f"{method} Pull Request"
-        elif "repos" in endpoint:
-            return "github", f"{method} Repository"
-        elif "search" in endpoint:
-            return "github", "Search GitHub"
-        return "github", f"{method} GitHub API"
-
-    return "search", func_name
-
-async def stream_chat_pipeline(request_obj, body: dict, http_client):
-    user_prompt = body.get("prompt", "")
-    req_messages = body.get("messages")
-    images = body.get("images", [])
-    user_id = body.get("user_id", "default_user")
-    connectors = body.get("connectors", [])
-    is_web_search = bool(body.get("web_search", False))
-
-    messages = build_context_messages(req_messages, user_prompt, images)
-    available_tools = get_filtered_tools(connectors, web_search=is_web_search)
-
-    headers = {
-        "Authorization": f"Bearer {BAI_API_KEY}",
-        "Content-Type": "application/json",
-        "Accept": "text/event-stream"
-    }
-
-    curr_messages = list(messages)
-    max_turns = 5
-    tool_step_counter = 0
-    search_calls_count = 0
-    executed_search_queries = {}
-    aggregated_sources = []
-
-    try:
-        for turn_index in range(max_turns):
-            active_tools = list(available_tools)
-            if search_calls_count >= 3:
-                active_tools = [t for t in active_tools if t.get("function", {}).get("name") != "tavily_search"]
-
-            payload = {
-                "model": MODEL_NAME,
-                "messages": curr_messages,
-                "temperature": 0.5,
-                "top_p": 0.95,
-                "max_tokens": 16384,
-                "stream": True
-            }
-
-            if active_tools:
-                payload["tools"] = active_tools
-                payload["tool_choice"] = "auto"
-
-            tool_calls_acc = []
-            has_tool_call = False
-
-            async with http_client.stream("POST", BAI_URL, headers=headers, json=payload) as response:
-                async for line in response.aiter_lines():
-                    if await request_obj.is_disconnected():
-                        return
-                    if not line:
-                        continue
-                    line_str = line.strip()
-                    if line_str.startswith("data: "):
-                        data_content = line_str[6:].strip()
-                        if data_content == "[DONE]":
-                            break
-                        try:
-                            json_data = json.loads(data_content)
-                            choices = json_data.get("choices", [])
-                            if choices:
-                                delta = choices[0].get("delta", {})
-                                reasoning = delta.get("reasoning_content") or delta.get("reasoning")
-                                content = delta.get("content")
-                                tc_delta = delta.get("tool_calls")
-
-                                if reasoning:
-                                    yield json.dumps({"thinking": reasoning}) + "\n"
-
-                                if tc_delta:
-                                    has_tool_call = True
-                                    for tc in tc_delta:
-                                        idx = tc.get("index", 0)
-                                        while len(tool_calls_acc) <= idx:
-                                            tool_calls_acc.append({"id": "", "type": "function", "function": {"name": "", "arguments": ""}})
-                                        if tc.get("id"):
-                                            tool_calls_acc[idx]["id"] += tc["id"]
-                                        if tc.get("function", {}).get("name"):
-                                            tool_calls_acc[idx]["function"]["name"] += tc["function"]["name"]
-                                        if tc.get("function", {}).get("arguments"):
-                                            tool_calls_acc[idx]["function"]["arguments"] += tc["function"]["arguments"]
-
-                                if content and not has_tool_call:
-                                    yield json.dumps({"answer": content}) + "\n"
-                        except json.JSONDecodeError:
-                            continue
-
-            valid_tool_calls = [tc for tc in tool_calls_acc if tc.get("function", {}).get("name")]
-            if has_tool_call and valid_tool_calls:
-                for i, tc in enumerate(valid_tool_calls):
-                    if not tc.get("id"):
-                        tc["id"] = f"call_{turn_index}_{i+1}"
-
-                curr_messages.append({
-                    "role": "assistant",
-                    "tool_calls": valid_tool_calls
-                })
-
-                for tc in valid_tool_calls:
-                    fn_name = tc.get("function", {}).get("name", "")
-                    args_raw = tc.get("function", {}).get("arguments", "{}")
-                    try:
-                        args_dict = json.loads(args_raw) if args_raw else {}
-                    except Exception:
-                        args_dict = {}
-
-                    tool_step_counter += 1
-                    icon_type, friendly_title = get_tool_metadata(fn_name, args_dict)
-
-                    step_add_payload = {
-                        "tool_event": "step_add",
-                        "step_index": tool_step_counter,
-                        "type": "tool_call",
-                        "name": fn_name,
-                        "title": friendly_title,
-                        "icon": icon_type,
-                        "request": args_dict,
-                        "response": ""
-                    }
-                    yield json.dumps(step_add_payload) + "\n"
-
-                    if fn_name == "tavily_search":
-                        q = args_dict.get("query", user_prompt or "").strip()
-                        depth = args_dict.get("search_depth", "basic")
-                        max_r = args_dict.get("max_results", 8)
-
-                        norm_q = normalize_query_for_cache(q)
-
-                        if norm_q in executed_search_queries:
-                            cached_res, cached_sources = executed_search_queries[norm_q]
-                            tool_res = cached_res
-                            sources_list = cached_sources
-                        else:
-                            search_calls_count += 1
-                            tool_res, sources_list = await execute_tavily_search(
-                                query=q,
-                                max_results=max_r,
-                                search_depth=depth
-                            )
-                            executed_search_queries[norm_q] = (tool_res, sources_list)
-
-                        if sources_list:
-                            for s in sources_list:
-                                if not any(existing.get("url") == s.get("url") for existing in aggregated_sources):
-                                    aggregated_sources.append(s)
-                            yield json.dumps({"sources": aggregated_sources}) + "\n"
-                    else:
-                        tool_res = await execute_tool_call(user_id, fn_name, args_dict)
-
-                    step_update_payload = {
-                        "tool_event": "step_result",
-                        "step_index": tool_step_counter,
-                        "response": str(tool_res)
-                    }
-                    yield json.dumps(step_update_payload) + "\n"
-
-                    curr_messages.append({
-                        "role": "tool",
-                        "tool_call_id": tc.get("id"),
-                        "content": str(tool_res)
-                    })
-            else:
-                break
-
-        if tool_step_counter > 0:
-            yield json.dumps({"tool_event": "finished", "total_steps": tool_step_counter}) + "\n"
-
-    except asyncio.CancelledError:
-        pass