diff --git a/app/api/v1/chat.py b/app/api/v1/chat.py
index 26924ec..81f9487 100644
--- a/app/api/v1/chat.py
+++ b/app/api/v1/chat.py
@@ -29,6 +29,13 @@ from app.services.grok.utils.response import (
)
+try:
+ with open("app_traffic.log", "w", encoding="utf-8") as f:
+ import datetime
+ f.write(f"=== app_traffic.log Cleared on Uvicorn Loaded [{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] ===\n\n")
+except Exception:
+ pass
+
class MessageItem(BaseModel):
"""消息项"""
@@ -393,7 +400,7 @@ def validate_request(request: ChatCompletionRequest):
# 字符串内容
if isinstance(content, str):
- if not content.strip():
+ if not content.strip() and msg.role != "assistant":
raise ValidationException(
message="Message content cannot be empty",
param=f"messages.{idx}.content",
@@ -514,7 +521,7 @@ def validate_request(request: ChatCompletionRequest):
"file.file_data",
f"messages.{idx}.content.{block_idx}.file.file_data",
)
- elif content is None and msg.role == "tool":
+ elif content is None and (msg.role == "tool" or msg.tool_calls):
pass
elif not isinstance(content, list):
raise ValidationException(
@@ -1034,6 +1041,20 @@ async def chat_completions(request: ChatCompletionRequest, raw_request: Request)
except Exception as e:
return _chat_error_as_success_response(request.model, _video_error_message(e))
else:
+ try:
+ with open("app_traffic.log", "a", encoding="utf-8") as f:
+ import datetime
+ f.write(f"--- Chat Request [{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] ---\n")
+ dump = {
+ "model": request.model,
+ "messages": [m.model_dump() for m in request.messages],
+ "stream": request.stream,
+ "tools": request.tools if request.tools else None,
+ }
+ f.write(orjson.dumps(dump, option=orjson.OPT_INDENT_2).decode('utf-8') + "\n\n")
+ except Exception:
+ pass
+
result = await ChatService.completions(
model=request.model,
messages=[msg.model_dump() for msg in request.messages],
diff --git a/app/services/grok/services/chat.py b/app/services/grok/services/chat.py
index cbfb6ba..9c94573 100644
--- a/app/services/grok/services/chat.py
+++ b/app/services/grok/services/chat.py
@@ -1,4 +1,4 @@
-"""
+"""
Grok Chat 服务
"""
@@ -130,7 +130,14 @@ class MessageExtractor:
if isinstance(content, str):
if content.strip():
- parts.append(content)
+ text_to_add = content
+ if role == "tool" and len(text_to_add) > 3000:
+ text_to_add = (
+ text_to_add[:1500]
+ + "\n\n... [Output Truncated for Conciseness] ...\n\n"
+ + text_to_add[-1000:]
+ )
+ parts.append(text_to_add)
elif isinstance(content, dict):
content = [content]
for item in content:
@@ -209,11 +216,11 @@ class MessageExtractor:
role_label = role
if role == "tool":
name = msg.get("name")
- call_id = msg.get("tool_call_id")
if isinstance(name, str) and name.strip():
- role_label = f"tool[{name.strip()}]"
- if isinstance(call_id, str) and call_id.strip():
- role_label = f"{role_label}#{call_id.strip()}"
+ role_label = f"--- Action Result: {name.strip()} ---"
+ full_text = "\n".join(parts)
+ if "exited with code" in full_text and "code 0" not in full_text:
+ role_label += " [FAILED]"
extracted.append({"role": role_label, "text": "\n".join(parts)})
# 找到最后一条 user 消息
@@ -239,7 +246,26 @@ class MessageExtractor:
if tools:
tool_prompt = build_tool_prompt(tools, tool_choice, parallel_tool_calls)
if tool_prompt:
- combined = f"{tool_prompt}\n\n{combined}"
+ # 注入 Few-Shot 引导以避免模型惯性,强化 Tool Calling 回语。
+ few_shot = (
+ "\n\n--- CRITICAL: TOOL CALL REQUIREMENT ---\n"
+ "Example:\n"
+ "User: test.txt 里面有什么内容?\n"
+ "Assistant: {\"name\": \"read\", \"arguments\": {\"path\": \"test.txt\"}}\n"
+ "User: 【System Notification: Tool Execution Result - read】\nhello world\n"
+ "Assistant: test.txt 的内容是 hello world。\n"
+ "--- END EXAMPLE ---\n\n"
+ "Important Guidelines:\n"
+ "1. **Task Complete**: If the information you have (including tool execution results) is SUFFICIENT to answer the user, you MUST answer directly in human language. DO NOT call any tools.\n"
+ "2. **More Tools Needed**: If it is INDISPENSABLE to use a tool, you MUST output ONLY the raw `` block without conversational narrative filler."
+ )
+ combined = f"{tool_prompt}{few_shot}\n\n{combined}"
+ # 尾插强化 (Tail Enforcement)
+ combined += (
+ "\n\n--- Important Note ---\n"
+ "- Only respond using the valid structural `` JSON tag if you need data from available tools.\n"
+ "- Provide only one single execution block and wait for response without dialogue."
+ )
return combined, file_attachments, image_attachments
diff --git a/app/services/grok/utils/tool_call.py b/app/services/grok/utils/tool_call.py
index 11fef2a..afcb8bb 100644
--- a/app/services/grok/utils/tool_call.py
+++ b/app/services/grok/utils/tool_call.py
@@ -34,9 +34,13 @@ def build_tool_prompt(
return ""
lines = [
+ "Important Rule: Focus strictly on answering using tool output data and JSON.",
+ "To satisfy the task, leverage available tools in sequence without adding conversational filler.",
+ "Do not formulate standard chatbot dialogue or greet.",
+ "",
"# Available Tools",
"",
- "You have access to the following tools. To call a tool, output a block with a JSON object containing \"name\" and \"arguments\".",
+ "To call a tool, you MUST output a block containing a JSON object with \"name\" and \"arguments\".",
"",
"Format:",
"",
@@ -77,10 +81,14 @@ def build_tool_prompt(
lines.append(f"IMPORTANT: You MUST call the tool \"{forced_name}\" in your response.")
else:
# "auto" or default
- lines.append("Decide whether to call a tool based on the user's request. If you don't need a tool, respond normally with text only.")
+ lines.append("Decide whether to call a tool based on the user's request. IF YOU CALL A TOOL, DO NOT ADD ANY DIALOGUE. OUTPUT FAST AND ONLY.")
lines.append("")
lines.append("When you call a tool, you may include text before or after the blocks, but the tool call blocks must be valid JSON.")
+ lines.append("")
+ lines.append("(If the task is complete and no more tools are needed, provide a regular answer. Only use when you actually need to call a tool.)")
+ lines.append("")
+ lines.append("IMPORTANT (Tool Error Handling): If a tool (e.g., `read`) fails with `EISDIR: illegal operation on a directory`, it means the path is a directory. You MUST call `exec` with `dir` (on Windows) or `ls` (on Linux) to list its contents instead.")
return "\n".join(lines)
@@ -296,14 +304,8 @@ def format_tool_history(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
})
elif role == "tool":
- # Convert tool result to text format
- tool_name = name or "unknown"
- call_id = tool_call_id or ""
- content_str = content if isinstance(content, str) else json.dumps(content, ensure_ascii=False) if content else ""
- result.append({
- "role": "user",
- "content": f"tool ({tool_name}, {call_id}): {content_str}",
- })
+ # 保留 tool 角色,交给 MessageExtractor 统一压平或由后台处理
+ result.append(msg)
else:
result.append(msg)
diff --git a/app/services/reverse/app_chat.py b/app/services/reverse/app_chat.py
index ac90b60..b255c3a 100644
--- a/app/services/reverse/app_chat.py
+++ b/app/services/reverse/app_chat.py
@@ -39,6 +39,18 @@ def _is_transient_network_error(err: Exception) -> bool:
class AppChatReverse:
"""/rest/app-chat/conversations/new reverse interface."""
+ @staticmethod
+ def _resolve_custom_personality() -> Optional[str]:
+ """Resolve optional custom personality from app config."""
+ value = get_config("app.custom_instruction", "")
+ if value is None:
+ return None
+ if not isinstance(value, str):
+ value = str(value)
+ if not value.strip():
+ return None
+ return value
+
@staticmethod
def build_payload(
message: str,
@@ -94,6 +106,13 @@ class AppChatReverse:
if model_config_override:
payload["responseMetadata"]["modelConfigOverride"] = model_config_override
+ if model == "grok-420":
+ payload["enable420"] = True
+
+ custom_personality = AppChatReverse._resolve_custom_personality()
+ if custom_personality is not None and "Greet the user" not in message[-1000:]:
+ payload["customPersonality"] = custom_personality
+
return payload
@staticmethod
diff --git a/app/services/reverse/utils/headers.py b/app/services/reverse/utils/headers.py
index b2e1aa3..9aa1c85 100644
--- a/app/services/reverse/utils/headers.py
+++ b/app/services/reverse/utils/headers.py
@@ -1,8 +1,8 @@
"""Shared header builders for reverse interfaces."""
+import re
import uuid
import orjson
-import random
from urllib.parse import urlparse
from typing import Dict, Optional
@@ -10,6 +10,52 @@ from app.core.logger import logger
from app.core.config import get_config
from app.services.reverse.utils.statsig import StatsigGenerator
+_HEADER_CHAR_REPLACEMENTS = str.maketrans(
+ {
+ "\u2010": "-", # hyphen
+ "\u2011": "-", # non-breaking hyphen
+ "\u2012": "-", # figure dash
+ "\u2013": "-", # en dash
+ "\u2014": "-", # em dash
+ "\u2212": "-", # minus sign
+ "\u2018": "'", # left single quote
+ "\u2019": "'", # right single quote
+ "\u201c": '"', # left double quote
+ "\u201d": '"', # right double quote
+ "\u00a0": " ", # nbsp
+ "\u2007": " ", # figure space
+ "\u202f": " ", # narrow nbsp
+ "\u200b": "", # zero width space
+ "\u200c": "", # zero width non-joiner
+ "\u200d": "", # zero width joiner
+ "\ufeff": "", # bom
+ }
+)
+
+
+def _sanitize_header_value(
+ value: Optional[str],
+ *,
+ field_name: str,
+ remove_all_spaces: bool = False,
+) -> str:
+ """Normalize header values and make sure they are latin-1 safe."""
+ raw = "" if value is None else str(value)
+ normalized = raw.translate(_HEADER_CHAR_REPLACEMENTS)
+ if remove_all_spaces:
+ normalized = re.sub(r"\s+", "", normalized)
+ else:
+ normalized = normalized.strip()
+
+ # curl_cffi header encoding defaults to latin-1.
+ normalized = normalized.encode("latin-1", errors="ignore").decode("latin-1")
+
+ if normalized != raw:
+ logger.warning(
+ f"Sanitized header field '{field_name}' (len {len(raw)} -> {len(normalized)})"
+ )
+ return normalized
+
def build_sso_cookie(sso_token: str) -> str:
"""
@@ -23,15 +69,41 @@ def build_sso_cookie(sso_token: str) -> str:
"""
# Format
sso_token = sso_token[4:] if sso_token.startswith("sso=") else sso_token
+ sso_token = _sanitize_header_value(
+ sso_token, field_name="sso_token", remove_all_spaces=True
+ )
# SSO Cookie
cookie = f"sso={sso_token}; sso-rw={sso_token}"
# CF Cookies
- cf_cookies = get_config("proxy.cf_cookies") or ""
- if not cf_cookies:
- cf_clearance = get_config("proxy.cf_clearance")
- if cf_clearance:
+ cf_cookies = _sanitize_header_value(
+ get_config("proxy.cf_cookies") or "", field_name="proxy.cf_cookies"
+ )
+ cf_clearance = _sanitize_header_value(
+ get_config("proxy.cf_clearance") or "",
+ field_name="proxy.cf_clearance",
+ remove_all_spaces=True,
+ )
+ cf_refresh_enabled = bool(get_config("proxy.enabled"))
+
+ if cf_refresh_enabled:
+ if not cf_cookies and cf_clearance:
+ cf_cookies = f"cf_clearance={cf_clearance}"
+ elif cf_clearance:
+ if cf_cookies:
+ # Replace existing cf_clearance or append if missing.
+ if re.search(r"(?:^|;\\s*)cf_clearance=", cf_cookies):
+ cf_cookies = re.sub(
+ r"(^|;\\s*)cf_clearance=[^;]*",
+ r"\\1cf_clearance=" + cf_clearance,
+ cf_cookies,
+ count=1,
+ )
+ else:
+ cf_cookies = cf_cookies.rstrip("; ")
+ cf_cookies = f"{cf_cookies}; cf_clearance={cf_clearance}"
+ else:
cf_cookies = f"cf_clearance={cf_clearance}"
if cf_cookies:
if cookie and not cookie.endswith(";"):
@@ -41,6 +113,95 @@ def build_sso_cookie(sso_token: str) -> str:
return cookie
+def _extract_major_version(browser: Optional[str], user_agent: Optional[str]) -> Optional[str]:
+ if browser:
+ match = re.search(r"(\d{2,3})", browser)
+ if match:
+ return match.group(1)
+ if user_agent:
+ for pattern in [r"Edg/(\d+)", r"Chrome/(\d+)", r"Chromium/(\d+)"]:
+ match = re.search(pattern, user_agent)
+ if match:
+ return match.group(1)
+ return None
+
+
+def _detect_platform(user_agent: str) -> Optional[str]:
+ ua = user_agent.lower()
+ if "windows" in ua:
+ return "Windows"
+ if "mac os x" in ua or "macintosh" in ua:
+ return "macOS"
+ if "android" in ua:
+ return "Android"
+ if "iphone" in ua or "ipad" in ua:
+ return "iOS"
+ if "linux" in ua:
+ return "Linux"
+ return None
+
+
+def _detect_arch(user_agent: str) -> Optional[str]:
+ ua = user_agent.lower()
+ if "aarch64" in ua or "arm" in ua:
+ return "arm"
+ if "x86_64" in ua or "x64" in ua or "win64" in ua or "intel" in ua:
+ return "x86"
+ return None
+
+
+def _build_client_hints(browser: Optional[str], user_agent: Optional[str]) -> Dict[str, str]:
+ browser = (browser or "").strip().lower()
+ user_agent = user_agent or ""
+ ua = user_agent.lower()
+
+ is_edge = "edge" in browser or "edg" in ua
+ is_brave = "brave" in browser
+ is_chromium = any(key in browser for key in ["chrome", "chromium", "edge", "brave"]) or (
+ "chrome" in ua or "chromium" in ua or "edg" in ua
+ )
+ is_firefox = "firefox" in ua or "firefox" in browser
+ is_safari = ("safari" in ua and "chrome" not in ua and "chromium" not in ua and "edg" not in ua) or "safari" in browser
+
+ if not is_chromium or is_firefox or is_safari:
+ return {}
+
+ version = _extract_major_version(browser, user_agent)
+ if not version:
+ return {}
+
+ if is_edge:
+ brand = "Microsoft Edge"
+ elif "chromium" in browser:
+ brand = "Chromium"
+ elif is_brave:
+ brand = "Brave"
+ else:
+ brand = "Google Chrome"
+
+ sec_ch_ua = (
+ f"\"{brand}\";v=\"{version}\", "
+ f"\"Chromium\";v=\"{version}\", "
+ "\"Not(A:Brand\";v=\"24\""
+ )
+
+ platform = _detect_platform(user_agent)
+ arch = _detect_arch(user_agent)
+ mobile = "?1" if ("mobile" in ua or platform in ("Android", "iOS")) else "?0"
+
+ hints = {
+ "Sec-Ch-Ua": sec_ch_ua,
+ "Sec-Ch-Ua-Mobile": mobile,
+ }
+ if platform:
+ hints["Sec-Ch-Ua-Platform"] = f"\"{platform}\""
+ if arch:
+ hints["Sec-Ch-Ua-Arch"] = arch
+ hints["Sec-Ch-Ua-Bitness"] = "64"
+ hints["Sec-Ch-Ua-Model"] = "" if mobile == "?0" else ""
+ return hints
+
+
def build_ws_headers(token: Optional[str] = None, origin: Optional[str] = None, extra: Optional[Dict[str, str]] = None) -> Dict[str, str]:
"""
Build headers for WebSocket requests.
@@ -53,12 +214,24 @@ def build_ws_headers(token: Optional[str] = None, origin: Optional[str] = None,
Returns:
Dict[str, str]: The headers dictionary.
"""
+ user_agent = _sanitize_header_value(
+ get_config("proxy.user_agent"), field_name="proxy.user_agent"
+ )
+ if not user_agent:
+ user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 Safari/537.36"
+ safe_origin = _sanitize_header_value(origin or "https://grok.com", field_name="origin")
headers = {
- "Origin": origin or "https://grok.com",
+ "Origin": safe_origin,
+ "User-Agent": user_agent,
+ "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Cache-Control": "no-cache",
"Pragma": "no-cache",
}
+ client_hints = _build_client_hints(get_config("proxy.browser"), user_agent)
+ if client_hints:
+ headers.update(client_hints)
+
if token:
headers["Cookie"] = build_sso_cookie(token)
@@ -81,22 +254,29 @@ def build_headers(cookie_token: str, content_type: Optional[str] = None, origin:
Returns:
Dict[str, str]: The headers dictionary.
"""
- trace_id = uuid.uuid4().hex
- span_id = uuid.uuid4().hex[:16]
-
+ user_agent = _sanitize_header_value(
+ get_config("proxy.user_agent"), field_name="proxy.user_agent"
+ )
+ if not user_agent:
+ user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 Safari/537.36"
+ safe_origin = _sanitize_header_value(origin or "https://grok.com", field_name="origin")
+ safe_referer = _sanitize_header_value(
+ referer or "https://grok.com/", field_name="referer"
+ )
headers = {
- "Baggage": f"sentry-environment=production,sentry-release=c43385ae21231a335971832ae5b5e8bdba69852d,sentry-public_key=b311e0f2690c81f25e2c4cf6d4f7ce1c,sentry-trace_id={trace_id},sentry-org_id=4508179396558848,sentry-sampled=false,sentry-sample_rand={random.random()},sentry-sample_rate=0",
- "Origin": origin or "https://grok.com",
+ "Accept-Encoding": "gzip, deflate, br, zstd",
+ "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
+ "Baggage": "sentry-environment=production,sentry-release=d6add6fb0460641fd482d767a335ef72b9b6abb8,sentry-public_key=b311e0f2690c81f25e2c4cf6d4f7ce1c",
+ "Origin": safe_origin,
"Priority": "u=1, i",
- "Referer": referer or "https://grok.com/",
+ "Referer": safe_referer,
"Sec-Fetch-Mode": "cors",
- "Sentry-Trace": f"{trace_id}-{span_id}-0",
- "Traceparent": f"00-{trace_id}-{span_id}-00",
+ "User-Agent": user_agent,
}
-
- user_agent = get_config("proxy.user_agent")
- if user_agent:
- headers["User-Agent"] = user_agent
+
+ client_hints = _build_client_hints(get_config("proxy.browser"), user_agent)
+ if client_hints:
+ headers.update(client_hints)
# Cookie
headers["Cookie"] = build_sso_cookie(cookie_token)