Commit 217c2686 authored by Kantz's avatar Kantz
Browse files

Toolcalling für CAS system hinzugefügt

parent aff17010
......@@ -7,6 +7,10 @@ DAILY_LLM_TOKEN_LIMIT="500000"
FRONTEND_URL="http://frontend:3000"
LLM_TOOL_USE_ENABLED="False" # "True" or "False"
MCP_SHARED_SECRET=""
MCP_BASE_URL=""
ORCHESTRATOR="task" # "tutor", "task" or "qa"
TASK_FOLDER="tasks"
LLM_PROVIDER="gwdg" # "openai", "gwdg", "mistral", or "ollama"
......
from __future__ import annotations
import asyncio
import logging
from threading import Lock
from typing import Any, Dict
......@@ -10,7 +11,7 @@ from fastapi import APIRouter
from fastapi.responses import JSONResponse
import app.config as config
from app.deterministic_services import llm_quota
from app.deterministic_services import llm_client, llm_quota
router = APIRouter()
logger = logging.getLogger(__name__)
......@@ -155,14 +156,42 @@ def _check_llm_quota() -> dict:
return {"status": "error", "detail": str(exc)}
async def _probe_mcp_server() -> dict:
mcp_settings = config.get_mcp_settings()
try:
url = llm_client._resolve_mcp_base_url(mcp_settings)
except ValueError as exc:
return {"status": "missing_config", "detail": str(exc)}
async def _handler(session, tools) -> dict:
return {
"status": "ok",
"url": url,
"tool_count": len(tools),
}
try:
return await llm_client._with_mcp_session(mcp_settings, _handler)
except Exception as exc:
return {"status": "error", "url": url, "detail": str(exc)}
def _check_mcp() -> dict:
if not config.get_llm_tool_use_enabled():
return {"status": "disabled"}
return asyncio.run(_probe_mcp_server())
@router.get("/api/health")
def health() -> Dict[str, Any]:
services = {
**_check_selected_llm_provider(),
"mcp": _check_mcp(),
"postgres": _check_postgres(),
"llm_quota": _check_llm_quota(),
}
required_statuses = {"ok"}
required_statuses = {"ok", "disabled"}
overall = "ok" if all(
service.get("status") in required_statuses for service in services.values()
) else "degraded"
......
......@@ -54,6 +54,11 @@ def get_llm_provider() -> str:
return provider
def get_llm_tool_use_enabled() -> bool:
value = os.getenv("LLM_TOOL_USE_ENABLED", "false")
normalized = value.strip().lower()
return normalized == "true"
def get_embedding_provider() -> str:
value = os.getenv("EMBEDDING_PROVIDER")
if value is not None and value.strip():
......@@ -117,6 +122,11 @@ def get_embedding_settings() -> EmbeddingSettings:
raise ValueError(f"Unsupported EMBEDDING_PROVIDER: {provider}")
def get_mcp_settings() -> dict:
return {
"base_url": os.getenv("MCP_BASE_URL"),
"shared_secret": os.getenv("MCP_SHARED_SECRET"),
}
@dataclass(frozen=True)
class OllamaSettings:
......
import asyncio
import inspect
import json
from datetime import date
from typing import Any, Callable
import httpx
import ollama
from openai import OpenAI
from mcp.client.session import ClientSession
from mcp.client.streamable_http import streamable_http_client
from mistralai.client import Mistral
from openai import AsyncOpenAI, OpenAI
from app import config
from app.deterministic_services import llm_quota
from app.deterministic_services import llm_quota, tool_log_context
AUTH_HEADER = "X-MCP-Shared-Secret"
# ---
......@@ -21,6 +28,7 @@ def _filter_kwargs(func, kwargs: dict) -> dict:
return kwargs
return {key: value for key, value in kwargs.items() if key in signature.parameters}
def _extract_message(response) -> object:
if isinstance(response, dict) and "message" in response:
return response["message"]
......@@ -29,6 +37,14 @@ def _extract_message(response) -> object:
return {}
def _extract_tool_calls(message: object) -> list[Any]:
if isinstance(message, dict):
return list(message.get("tool_calls") or [])
if hasattr(message, "tool_calls"):
return list(getattr(message, "tool_calls") or [])
return []
def _extract_total_tokens(response: object) -> int:
usage = None
if isinstance(response, dict):
......@@ -55,6 +71,91 @@ def _extract_total_tokens(response: object) -> int:
return int(prompt) + int(completion)
def _normalize_message(message: object) -> dict[str, Any]:
if isinstance(message, dict):
return dict(message)
normalized: dict[str, Any] = {}
for attr in ("role", "content", "tool_calls", "name"):
if hasattr(message, attr):
value = getattr(message, attr)
if value is not None:
normalized[attr] = value
return normalized
def _extract_tool_call_name(tool_call: object) -> str:
function = getattr(tool_call, "function", None)
if function is not None and hasattr(function, "name"):
return getattr(function, "name")
if isinstance(tool_call, dict):
function_dict = tool_call.get("function") or {}
return str(function_dict.get("name") or "")
return ""
def _extract_tool_call_arguments(tool_call: object) -> dict[str, Any]:
function = getattr(tool_call, "function", None)
raw_arguments: Any = None
if function is not None and hasattr(function, "arguments"):
raw_arguments = getattr(function, "arguments")
elif isinstance(tool_call, dict):
function_dict = tool_call.get("function") or {}
raw_arguments = function_dict.get("arguments")
if isinstance(raw_arguments, dict):
return raw_arguments
if isinstance(raw_arguments, str) and raw_arguments.strip():
return json.loads(raw_arguments)
return {}
def _extract_tool_call_id(tool_call: object) -> str:
if hasattr(tool_call, "id"):
return str(getattr(tool_call, "id") or "")
if isinstance(tool_call, dict):
return str(tool_call.get("id") or "")
return ""
def _extract_structured_tool_output(tool_result: Any) -> Any:
if hasattr(tool_result, "structured_content") and tool_result.structured_content is not None:
return tool_result.structured_content
if hasattr(tool_result, "structuredContent") and tool_result.structuredContent is not None:
return tool_result.structuredContent
blocks = []
for item in getattr(tool_result, "content", []) or []:
text = getattr(item, "text", None)
if text is not None:
blocks.append(text)
return "\n".join(blocks)
def _append_tool_trace_log(result: dict, provider: str) -> None:
tool_trace = result.get("tool_trace")
if not tool_trace:
return
tool_log = tool_log_context.get_active_tool_log()
if tool_log is None:
return
tool_log.append(
{
"name": "llm_tool_trace",
"arguments": {
"provider": provider,
"tool_call_count": len(tool_trace),
},
"response": {
"tool_trace": tool_trace,
"response_ids": result.get("response_ids", []),
},
}
)
# ---
# Quota
# ---
......@@ -125,7 +226,7 @@ def _require_mistral_chat_settings() -> config.MistralChatSettings:
# ---
# API calls
# Plain API calls
# ---
def _chat_openai_compatible(
......@@ -134,8 +235,7 @@ def _chat_openai_compatible(
) -> dict:
settings = settings or _require_openai_chat_settings()
timeout = settings.timeout or 60.0
client = OpenAI(api_key=settings.api_key,
base_url=settings.base_url, timeout=timeout)
client = OpenAI(api_key=settings.api_key, base_url=settings.base_url, timeout=timeout)
kwargs = {"messages": messages, "model": settings.model}
if settings.temperature is not None:
kwargs["temperature"] = settings.temperature
......@@ -184,6 +284,260 @@ def _chat_ollama(
return {"raw": response, "message": _extract_message(response)}
# ---
# MCP tool execution
# ---
def _resolve_mcp_base_url(mcp_settings: dict[str, Any]) -> str:
base_url = mcp_settings.get("base_url")
if not base_url:
raise ValueError("MCP_BASE_URL must be set")
return str(base_url)
async def _with_mcp_session(
mcp_settings: dict[str, Any],
handler,
http_client: httpx.AsyncClient | None = None,
):
headers: dict[str, str] = {}
shared_secret = mcp_settings.get("shared_secret")
if shared_secret:
headers[AUTH_HEADER] = str(shared_secret)
managed_http_client = http_client
if managed_http_client is None:
managed_http_client = httpx.AsyncClient(headers=headers, timeout=30.0)
elif headers:
managed_http_client.headers.update(headers)
should_close_http_client = http_client is None
try:
async with streamable_http_client(
_resolve_mcp_base_url(mcp_settings),
http_client=managed_http_client,
) as streams:
async with ClientSession(streams[0], streams[1]) as session:
await session.initialize()
listed_tools = await session.list_tools()
return await handler(session, listed_tools.tools)
finally:
if should_close_http_client:
await managed_http_client.aclose()
def _tool_attr(tool: Any, snake_name: str, camel_name: str) -> Any:
if hasattr(tool, snake_name):
return getattr(tool, snake_name)
return getattr(tool, camel_name)
def _tool_to_openai_chat_tool(tool: Any) -> dict[str, Any]:
return {
"type": "function",
"function": {
"name": _tool_attr(tool, "name", "name"),
"description": _tool_attr(tool, "description", "description") or "",
"parameters": _tool_attr(tool, "input_schema", "inputSchema"),
},
}
def _tool_to_ollama_chat_tool(tool: Any) -> dict[str, Any]:
return {
"type": "function",
"function": {
"name": _tool_attr(tool, "name", "name"),
"description": _tool_attr(tool, "description", "description") or "",
"parameters": _tool_attr(tool, "input_schema", "inputSchema"),
},
}
def _append_openai_tool_message(
messages: list[dict[str, Any]],
tool_call_id: str,
tool_name: str,
structured_output: Any,
) -> None:
messages.append(
{
"role": "tool",
"tool_call_id": tool_call_id,
"name": tool_name,
"content": json.dumps(structured_output),
}
)
def _append_ollama_tool_message(
messages: list[dict[str, Any]],
tool_name: str,
structured_output: Any,
) -> None:
messages.append(
{
"role": "tool",
"tool_name": tool_name,
"content": json.dumps(structured_output),
}
)
async def _run_openai_compatible_tool_loop(
messages: list[dict],
settings: config.OpenAIChatSettings,
http_client: httpx.AsyncClient | None = None,
openai_client: Any | None = None,
) -> dict:
mcp_settings = config.get_mcp_settings()
client_kwargs: dict[str, Any] = {"api_key": settings.api_key}
if settings.base_url:
client_kwargs["base_url"] = settings.base_url
if settings.timeout is not None:
client_kwargs["timeout"] = settings.timeout
client = openai_client or AsyncOpenAI(**client_kwargs)
async def _handler(session, tools) -> dict:
chat_tools = [_tool_to_openai_chat_tool(tool) for tool in tools]
request_messages = [dict(message) for message in messages]
response_ids: list[str] = []
tool_trace: list[dict[str, Any]] = []
while True:
request_kwargs: dict[str, Any] = {
"model": settings.model,
"messages": request_messages,
"tools": chat_tools,
}
if settings.temperature is not None:
request_kwargs["temperature"] = settings.temperature
response = await client.chat.completions.create(**request_kwargs)
response_id = getattr(response, "id", "")
if response_id:
response_ids.append(response_id)
message = response.choices[0].message if response.choices else {}
tool_calls = _extract_tool_calls(message)
if not tool_calls:
return {
"raw": response,
"message": message,
"tool_trace": tool_trace,
"response_ids": response_ids,
}
request_messages.append(_normalize_message(message))
for tool_call in tool_calls:
tool_name = _extract_tool_call_name(tool_call)
arguments = _extract_tool_call_arguments(tool_call)
tool_result = await session.call_tool(tool_name, arguments=arguments)
structured_output = _extract_structured_tool_output(tool_result)
tool_trace.append(
{
"name": tool_name,
"arguments": arguments,
"output": structured_output,
}
)
_append_openai_tool_message(
request_messages,
_extract_tool_call_id(tool_call),
tool_name,
structured_output,
)
return await _with_mcp_session(mcp_settings, _handler, http_client=http_client)
async def _run_ollama_tool_loop(
messages: list[dict],
http_client: httpx.AsyncClient | None = None,
ollama_client: Any | None = None,
) -> dict:
settings = config.get_ollama_settings()
mcp_settings = config.get_mcp_settings()
client_kwargs: dict[str, Any] = {"host": settings.base_url}
if settings.timeout is not None:
client_kwargs["timeout"] = settings.timeout
client = ollama_client or ollama.AsyncClient(**client_kwargs)
async def _handler(session, tools) -> dict:
chat_tools = [_tool_to_ollama_chat_tool(tool) for tool in tools]
request_messages = [dict(message) for message in messages]
response_ids: list[str] = []
tool_trace: list[dict[str, Any]] = []
while True:
request_kwargs: dict[str, Any] = {
"model": settings.model,
"messages": request_messages,
"tools": chat_tools,
}
if settings.keepalive is not None:
request_kwargs["keep_alive"] = settings.keepalive
if settings.temperature is not None:
request_kwargs["options"] = {"temperature": settings.temperature}
completion = await client.chat(**request_kwargs)
completion_id = getattr(completion, "id", "")
if completion_id:
response_ids.append(completion_id)
message = _extract_message(completion)
tool_calls = _extract_tool_calls(message)
if not tool_calls:
return {
"raw": completion,
"message": message,
"tool_trace": tool_trace,
"response_ids": response_ids,
}
request_messages.append(_normalize_message(message))
for tool_call in tool_calls:
tool_name = _extract_tool_call_name(tool_call)
arguments = _extract_tool_call_arguments(tool_call)
tool_result = await session.call_tool(tool_name, arguments=arguments)
structured_output = _extract_structured_tool_output(tool_result)
tool_trace.append(
{
"name": tool_name,
"arguments": arguments,
"output": structured_output,
}
)
_append_ollama_tool_message(request_messages, tool_name, structured_output)
return await _with_mcp_session(mcp_settings, _handler, http_client=http_client)
def _run_tool_chat_sync(
provider: str,
messages: list[dict],
settings: config.OpenAIChatSettings | None = None,
) -> dict:
if provider in {"openai", "gwdg"}:
assert settings is not None
return asyncio.run(_run_openai_compatible_tool_loop(messages, settings))
if provider == "ollama":
return asyncio.run(_run_ollama_tool_loop(messages))
raise ValueError(f"Unsupported tool-enabled LLM_PROVIDER: {provider}")
def _run_provider_chat(
provider: str,
chat_func: Callable[[], dict],
) -> dict:
result = _quota_tracked_chat(chat_func)
_append_tool_trace_log(result, provider)
return result
# ---
# Public chat API
# ---
......@@ -201,15 +555,29 @@ def chat(
messages: list[dict],
) -> dict:
provider = config.get_llm_provider()
tool_use_enabled = config.get_llm_tool_use_enabled()
if provider == "openai":
settings = _require_openai_chat_settings()
return _quota_tracked_chat(lambda: _chat_openai_compatible(messages, settings))
if tool_use_enabled:
return _run_provider_chat(
provider,
lambda: _run_tool_chat_sync(provider, messages, settings),
)
return _run_provider_chat(provider, lambda: _chat_openai_compatible(messages, settings))
if provider == "gwdg":
settings = _require_gwdg_chat_settings()
return _quota_tracked_chat(lambda: _chat_openai_compatible(messages, settings))
if tool_use_enabled:
return _run_provider_chat(
provider,
lambda: _run_tool_chat_sync(provider, messages, settings),
)
return _run_provider_chat(provider, lambda: _chat_openai_compatible(messages, settings))
if provider == "mistral":
settings = _require_mistral_chat_settings()
return _quota_tracked_chat(lambda: _chat_mistral(messages, settings))
return _run_provider_chat(provider, lambda: _chat_mistral(messages, settings))
if provider == "ollama":
return _chat_ollama(messages)
if tool_use_enabled:
return _run_provider_chat(provider, lambda: _run_tool_chat_sync(provider, messages))
return _run_provider_chat(provider, lambda: _chat_ollama(messages))
raise ValueError(f"Unsupported LLM_PROVIDER: {provider}")
......@@ -13,6 +13,7 @@ from app.deterministic_services import (
embedding_provider,
referenz_decoder,
retrieval_store,
tool_log_context,
tool_logging,
)
......@@ -257,25 +258,29 @@ def run_chat_common(
duration_ms=init_duration_ms,
)
if state.new_chat or not state.sheet.get("initialized"):
token = tool_log_context.set_active_tool_log(state.tool_log)
try:
if state.new_chat or not state.sheet.get("initialized"):
log_timed_call(
state.tool_log,
"orchestrator_bootstrap",
{"query": state.last_user},
lambda: on_bootstrap(state, state.last_user),
)
state.sheet["initialized"] = True
log_timed_call(
state.tool_log,
"orchestrator_bootstrap",
{"query": state.last_user},
lambda: on_bootstrap(state, state.last_user),
"orchestrator_turn_logic",
{"new_chat": state.new_chat},
lambda: on_turn_logic(state),
)
state.sheet["initialized"] = True
log_timed_call(
state.tool_log,
"orchestrator_turn_logic",
{"new_chat": state.new_chat},
lambda: on_turn_logic(state),
)
reply = log_timed_call(
state.tool_log,
"orchestrator_build_reply",
{"chat_id": state.chat_id},
lambda: on_build_reply(state),
)
return finalize_response(state, reply)
reply = log_timed_call(
state.tool_log,
"orchestrator_build_reply",
{"chat_id": state.chat_id},
lambda: on_build_reply(state),
)
return finalize_response(state, reply)
finally:
tool_log_context.reset_active_tool_log(token)
from __future__ import annotations
from contextvars import ContextVar, Token
_ACTIVE_TOOL_LOG: ContextVar[list[dict] | None] = ContextVar("active_tool_log", default=None)
def set_active_tool_log(tool_log: list[dict] | None) -> Token:
return _ACTIVE_TOOL_LOG.set(tool_log)
def reset_active_tool_log(token: Token) -> None:
_ACTIVE_TOOL_LOG.reset(token)
def get_active_tool_log() -> list[dict] | None:
return _ACTIVE_TOOL_LOG.get()
......@@ -59,6 +59,9 @@ class HealthReadinessUnitTest(unittest.TestCase):
) as ollama_check, patch(
"app.api.health._check_mistral"
) as mistral_check, patch(
"app.api.health._check_mcp",
return_value={"status": "disabled"},
) as mcp_check, patch(
"app.api.health._check_postgres",
return_value={"status": "ok"},
), patch(
......@@ -70,9 +73,10 @@ class HealthReadinessUnitTest(unittest.TestCase):
self.assertEqual(response["status"], "ok")
self.assertEqual(
set(response["services"].keys()),
{"gwdg", "postgres", "llm_quota"},
{"gwdg", "mcp", "postgres", "llm_quota"},
)
gwdg_check.assert_called_once()
mcp_check.assert_called_once()
openai_check.assert_not_called()
ollama_check.assert_not_called()
mistral_check.assert_not_called()
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment