feat(bot): GigaChat function-calling provider for the chat agent; never 500 on provider errors
CI / ruff · pytest (pull_request) Successful in 2m15s

OpenRouter returns 403 «Access denied by security policy» from the RF server, and parsing that string error
crashed the request (500 instead of the fallback answer). Error parsing is robust now and any agent exception
degrades to the extractive answer. GigaChat tool provider (GigaChat-2-Pro by default) converts the agent's
OpenAI-style history to function_call / role=function; auto mode prefers GigaChat when credentials are set.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-04 15:46:21 +03:00
co-authored by Claude Opus 5.5
parent bc044650ac
commit ade93d959b
4 changed files with 169 additions and 8 deletions
+86 -3
View File
@@ -68,9 +68,10 @@ class OpenRouterProvider:
if resp.status_code != 200:
# Provider error bodies are generic (no patient text) — log the message to make failures diagnosable.
try:
detail = str(resp.json().get("error", {}).get("message", ""))[:200]
except ValueError:
detail = ""
err = resp.json().get("error", "")
detail = str(err.get("message", "") if isinstance(err, dict) else err)[:200]
except (ValueError, AttributeError):
detail = resp.text[:200]
log.error("openrouter http %s: %s", resp.status_code, detail)
raise ProviderError(f"http {resp.status_code}")
try:
@@ -90,6 +91,88 @@ class OpenRouterProvider:
return AssistantTurn(content=msg.get("content"), tool_calls=calls)
class GigaChatToolProvider:
"""GigaChat function calling (RF provider, 152-FZ compliant). Converts the OpenAI-style history the agent keeps:
assistant `tool_calls` → `function_call`, `tool` results → role `function` with the function name.
GigaChat returns at most one function call per turn — the agent loop handles that naturally."""
name = "gigachat"
def __init__(self, settings: Settings, client: Any | None = None) -> None:
self._settings = settings
self._client = client
def _get_client(self) -> Any:
if self._client is None:
from gigachat import GigaChat
s = self._settings
self._client = GigaChat(
credentials=s.gigachat_credentials,
scope=s.gigachat_scope,
model=s.gigachat_agent_model,
verify_ssl_certs=s.gigachat_verify_ssl,
timeout=s.chat_agent_timeout_s,
)
return self._client
@staticmethod
def to_messages(messages: list[dict[str, Any]]) -> list[Any]:
from gigachat.models import FunctionCall, Messages, MessagesRole
out = []
for m in messages:
role = m["role"]
if role == "system":
out.append(Messages(role=MessagesRole.SYSTEM, content=m["content"]))
elif role == "user":
out.append(Messages(role=MessagesRole.USER, content=m["content"]))
elif role == "assistant":
calls = m.get("tool_calls") or []
for c in calls or [None]:
fc = None
if c is not None:
fc = FunctionCall(name=c["function"]["name"], arguments=json.loads(c["function"]["arguments"]))
out.append(Messages(role=MessagesRole.ASSISTANT, content=m.get("content") or "", function_call=fc))
elif role == "tool":
out.append(Messages(role=MessagesRole.FUNCTION, name=m.get("name"), content=m["content"]))
return out
@staticmethod
def to_functions(tools: list[dict[str, Any]]) -> list[Any]:
from gigachat.models import Function
return [Function.model_validate(t["function"]) for t in tools]
async def chat(self, messages: list[dict[str, Any]], tools: list[dict[str, Any]]) -> AssistantTurn:
from gigachat.models import Chat
chat = Chat(
messages=self.to_messages(messages),
functions=self.to_functions(tools),
function_call="auto",
temperature=0.1,
max_tokens=900,
)
try:
resp = await self._get_client().achat(chat)
except Exception as exc: # noqa: BLE001 — network/auth/HTTP errors of the SDK
raise ProviderError(type(exc).__name__) from exc
msg = resp.choices[0].message
calls = []
if msg.function_call is not None:
args = msg.function_call.arguments or {}
if isinstance(args, str):
try:
args = json.loads(args)
except json.JSONDecodeError:
args = {}
calls.append(ToolCall(id=msg.function_call.name, name=msg.function_call.name, arguments=args))
usage = getattr(resp, "usage", None)
log.info("gigachat agent ok calls=%d tokens=%s", len(calls), getattr(usage, "total_tokens", None))
return AssistantTurn(content=msg.content or None, tool_calls=calls)
# ------------------------------------------------------------------------------------------------ offline mock
_DRUG_WORDS = re.compile(