ed/manual_slop
Private
Public Access
0
0
Fork 0
Code Activity

refactor(ai_client): _send_grok_result() returns Result[str]

Browse Source
This commit is contained in:
ed
2026-06-12 18:43:47 -04:00
parent 49923f9b43
commit 87cac3808f
1 changed files with 40 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ai_client.py
+40 -37
View File
@@ -2247,50 +2247,53 @@ def _ensure_grok_client() -> Any:
_grok_client = openai.OpenAI(api_key=api_key, base_url="https://api.x.ai/v1") _grok_client = openai.OpenAI(api_key=api_key, base_url="https://api.x.ai/v1")
return _grok_client return _grok_client
def _send_grok(md_content: str, user_message: str, base_dir: str, def _send_grok_result(md_content: str, user_message: str, base_dir: str,
file_items: list[dict[str, Any]] | None = None, file_items: list[dict[str, Any]] | None = None,
discussion_history: str = "", discussion_history: str = "",
stream: bool = False, stream: bool = False,
pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None,
qa_callback: Optional[Callable[[str], str]] = None, qa_callback: Optional[Callable[[str], str]] = None,
stream_callback: Optional[Callable[[str], None]] = None, stream_callback: Optional[Callable[[str], None]] = None,
patch_callback: Optional[Callable[[str, str], Optional[str]]] = None) -> str: patch_callback: Optional[Callable[[str, str], Optional[str]]] = None) -> Result[str]:
from src.openai_compatible import OpenAICompatibleRequest from src.openai_compatible import OpenAICompatibleRequest, _classify_openai_compatible_error
client = _ensure_grok_client() try:
tools: list[dict[str, Any]] | None = _get_deepseek_tools() or None client = _ensure_grok_client()
caps = get_capabilities("grok", _model) tools: list[dict[str, Any]] | None = _get_deepseek_tools() or None
with _grok_history_lock: caps = get_capabilities("grok", _model)
user_content = user_message
if file_items:
for fi in file_items:
if fi.get("is_image") and fi.get("base64_data"):
user_content = f"[IMAGE: {fi.get('path', 'attachment')}]\n{user_content}"
if discussion_history and not _grok_history:
_grok_history.append({"role": "user", "content": f"[DISCUSSION HISTORY]\n\n{discussion_history}\n\n---\n\n{user_message}"})
else:
_grok_history.append({"role": "user", "content": user_content})
def _build_grok_request(_round_idx: int) -> OpenAICompatibleRequest:
with _grok_history_lock: with _grok_history_lock:
messages: list[dict[str, Any]] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>"}] user_content = user_message
messages.extend(_grok_history) if file_items:
extra_body: dict[str, Any] = {} for fi in file_items:
if caps.web_search: if fi.get("is_image") and fi.get("base64_data"):
extra_body["search_parameters"] = {"mode": "auto"} user_content = f"[IMAGE: {fi.get('path', 'attachment')}]\n{user_content}"
if caps.x_search: if discussion_history and not _grok_history:
extra_body.setdefault("search_parameters", {}) _grok_history.append({"role": "user", "content": f"[DISCUSSION HISTORY]\n\n{discussion_history}\n\n---\n\n{user_message}"})
extra_body["search_parameters"]["sources"] = [{"type": "x"}] else:
return OpenAICompatibleRequest( _grok_history.append({"role": "user", "content": user_content})
messages=messages, model=_model, temperature=_temperature, top_p=_top_p, def _build_grok_request(_round_idx: int) -> OpenAICompatibleRequest:
max_tokens=_max_tokens, stream=stream, stream_callback=stream_callback, with _grok_history_lock:
tools=tools, tool_choice="auto" if tools else "auto", messages: list[dict[str, Any]] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>"}]
extra_body=extra_body or None, messages.extend(_grok_history)
) extra_body: dict[str, Any] = {}
return run_with_tool_loop( if caps.web_search:
client, _build_grok_request, capabilities=caps, extra_body["search_parameters"] = {"mode": "auto"}
pre_tool_callback=pre_tool_callback, qa_callback=qa_callback, stream_callback=stream_callback, if caps.x_search:
patch_callback=patch_callback, base_dir=base_dir, vendor_name="grok", extra_body.setdefault("search_parameters", {})
history_lock=_grok_history_lock, history=_grok_history, extra_body["search_parameters"]["sources"] = [{"type": "x"}]
) return OpenAICompatibleRequest(
messages=messages, model=_model, temperature=_temperature, top_p=_top_p,
max_tokens=_max_tokens, stream=stream, stream_callback=stream_callback,
tools=tools, tool_choice="auto" if tools else "auto",
extra_body=extra_body or None,
)
return Result(data=run_with_tool_loop(
client, _build_grok_request, capabilities=caps,
pre_tool_callback=pre_tool_callback, qa_callback=qa_callback, stream_callback=stream_callback,
patch_callback=patch_callback, base_dir=base_dir, vendor_name="grok",
history_lock=_grok_history_lock, history=_grok_history,
))
except Exception as exc:
return Result(data="", errors=[_classify_openai_compatible_error(exc, source="ai_client.grok")])
def _list_grok_models() -> list[str]: def _list_grok_models() -> list[str]:
from src.vendor_capabilities import list_models_for_vendor from src.vendor_capabilities import list_models_for_vendor