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

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

Browse Source
This commit is contained in:
Edward R. Gonzalez
2026-06-12 18:47:14 -04:00
parent 6665152950
commit d4d7d1ab14
1 changed files with 27 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ai_client.py
+27 -24
View File
@@ -2536,36 +2536,39 @@ def ollama_chat(
resp = requests.post(f"{base_url}/api/chat", json=payload, timeout=120) resp = requests.post(f"{base_url}/api/chat", json=payload, timeout=120)
return resp.json() return resp.json()
def _send_llama_native(md_content: str, user_message: str, base_dir: str, def _send_llama_native_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]:
base_url = _llama_base_url.replace("/v1", "") try:
with _llama_history_lock: base_url = _llama_base_url.replace("/v1", "")
if discussion_history and not _llama_history: with _llama_history_lock:
_llama_history.append({"role": "user", "content": f"[DISCUSSION HISTORY]\n\n{discussion_history}\n\n---\n\n{user_message}"}) if discussion_history and not _llama_history:
else: _llama_history.append({"role": "user", "content": f"[DISCUSSION HISTORY]\n\n{discussion_history}\n\n---\n\n{user_message}"})
_llama_history.append({"role": "user", "content": user_message}) else:
messages: list[dict[str, Any]] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>"}] _llama_history.append({"role": "user", "content": user_message})
messages.extend(_llama_history) messages: list[dict[str, Any]] = [{"role": "system", "content": f"{_get_combined_system_prompt()}\n\n<context>\n{md_content}\n</context>"}]
images: list[str] = [] messages.extend(_llama_history)
if file_items: images: list[str] = []
for fi in file_items: if file_items:
if fi.get("is_image") and fi.get("base64_data"): for fi in file_items:
images.append(fi["base64_data"]) if fi.get("is_image") and fi.get("base64_data"):
response = ollama_chat(_model, messages, images=images, base_url=base_url) images.append(fi["base64_data"])
text = response.get("message", {}).get("content", "") response = ollama_chat(_model, messages, images=images, base_url=base_url)
thinking = response.get("message", {}).get("thinking", "") text = response.get("message", {}).get("content", "")
with _llama_history_lock: thinking = response.get("message", {}).get("thinking", "")
msg: dict[str, Any] = {"role": "assistant", "content": text or None} with _llama_history_lock:
if thinking: msg: dict[str, Any] = {"role": "assistant", "content": text or None}
msg["thinking"] = thinking if thinking:
_llama_history.append(msg) msg["thinking"] = thinking
return (f"<thinking>\n{thinking}\n</thinking>\n" if thinking else "") + text _llama_history.append(msg)
return Result(data=(f"<thinking>\n{thinking}\n</thinking>\n" if thinking else "") + text)
except Exception as exc:
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(exc), source="ai_client.llama_native", original=exc)])
def _list_llama_models() -> list[str]: def _list_llama_models() -> list[str]:
from src.vendor_capabilities import list_models_for_vendor from src.vendor_capabilities import list_models_for_vendor
return list_models_for_vendor("llama") return list_models_for_vendor("llama")