sources/webrtc-reviewed/src_llm/clients/local_chat.py

Текстовый файл из учебного комплекта, не свидетельство установленного runtime. Вернуться к учебнику · Исходный файл

   1 import json
   2 import re
   3 from typing import Any
   4 
   5 TOOL_CALL_RE = re.compile(r"<tool_call>\s*(.*?)\s*</tool_call>", re.S)
   6 
   7 
   8 def chat(
   9     llama,
  10     *,
  11     model: str,
  12     messages: list,
  13     tools: list,
  14     options: dict,
  15     max_tokens: int,
  16 ) -> dict[str, Any]:
  17     params = {
  18         "temperature": options.get("temperature", 0.0),
  19         "top_p": options.get("top_p"),
  20         "top_k": options.get("top_k"),
  21         "max_tokens": options.get("num_predict") or options.get("max_tokens") or max_tokens,
  22     }
  23     params = {k: v for k, v in params.items() if v is not None}
  24 
  25     response = llama.create_chat_completion(
  26         messages=messages,
  27         tools=tools or [],
  28         **params,
  29     )
  30     choices = response.get("choices") or []
  31     message = choices[0].get("message", {}) if choices else {}
  32     tool_calls = _parse_tool_calls(message.get("content") or "")
  33     if tool_calls:
  34         message["tool_calls"] = tool_calls
  35     _normalize_tool_calls(message)
  36     return {
  37         "message": message,
  38         "model": model,
  39         "provider": "local",
  40     }
  41 
  42 
  43 def _parse_tool_calls(content: str) -> list[dict]:
  44     calls = []
  45     for raw in TOOL_CALL_RE.findall(content):
  46         try:
  47             payload = json.loads(raw.strip())
  48         except json.JSONDecodeError:
  49             continue
  50         name = payload.get("name")
  51         if not name:
  52             continue
  53         calls.append({"function": {"name": name, "arguments": payload.get("arguments", {})}})
  54     return calls
  55 
  56 
  57 def _normalize_tool_calls(message: dict) -> None:
  58     for call in message.get("tool_calls") or []:
  59         function = call.get("function") or {}
  60         args = function.get("arguments")
  61         if isinstance(args, str):
  62             try:
  63                 function["arguments"] = json.loads(args)
  64             except json.JSONDecodeError:
  65                 function["arguments"] = {}