102 lines
4.2 KiB
Python
102 lines
4.2 KiB
Python
"""Backend profiles (D2): the seam between the framework and a MAF chat client.
|
|
|
|
A MAF agent binds to a *chat client* and the model is a parameter on that client — so model
|
|
choice is per-client (and per-agent via a role->deployment model-map, B12). A "backend
|
|
profile" selects how models are served and produces the corresponding MAF chat client.
|
|
|
|
Two GA-wired profiles (Fase 2):
|
|
|
|
* **LOCAL** (dev default, D6): ``OpenAIChatCompletionClient`` against an OpenAI-compatible
|
|
local endpoint (Ollama/LM Studio). Chat Completions, **non-streaming** — NOT the
|
|
Responses-based ``OpenAIChatClient`` (research 03: non-streaming populates ``UsageDetails``
|
|
None-safely and avoids the ``/v1`` tool-drop). The base URL defaults to loopback; no egress.
|
|
* **AZURE**: ``FoundryChatClient`` against a Foundry project (deployment names tenant-specific,
|
|
supplied via env + ``data/model_map.json``). Reserved for targeted, minimal verification.
|
|
|
|
``get_backend()`` and ``create_chat_client()`` are fail-fast (``ValueError``).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from enum import Enum
|
|
from importlib.resources import files
|
|
from typing import Protocol, runtime_checkable
|
|
|
|
from agent_framework import BaseChatClient
|
|
from agent_framework_foundry import FoundryChatClient
|
|
from agent_framework_openai import OpenAIChatCompletionClient
|
|
|
|
_MODEL_MAP_RESOURCE = "data/model_map.json"
|
|
# Loopback only — never a remote host (D6 / research 03 no-egress). Override via env.
|
|
_DEFAULT_LOCAL_BASE_URL = "http://127.0.0.1:11434/v1"
|
|
|
|
|
|
class Profile(str, Enum):
|
|
"""Model-serving backend profile (D2)."""
|
|
|
|
AZURE = "azure" # Foundry / Azure OpenAI — full profile
|
|
LOCAL = "local" # OpenAI-compatible local endpoint — fallback / dev default
|
|
|
|
|
|
@runtime_checkable
|
|
class ChatBackend(Protocol):
|
|
"""The seam: a backend produces a MAF chat client for a given model."""
|
|
|
|
profile: Profile
|
|
|
|
def create_chat_client(self, *, model: str) -> BaseChatClient:
|
|
"""Create a MAF chat client bound to ``model`` (the resolved deployment/model id)."""
|
|
...
|
|
|
|
|
|
def resolve_model(profile: Profile | str, role: str) -> str:
|
|
"""Resolve a role -> model/deployment id from ``data/model_map.json`` (B12). Falls back
|
|
to the profile's ``default``; fail-fast (``ValueError``) when nothing maps."""
|
|
prof = Profile(profile)
|
|
table = json.loads(
|
|
files("portfolio_optimiser").joinpath(_MODEL_MAP_RESOURCE).read_text(encoding="utf-8")
|
|
)
|
|
profile_map = table.get(prof.value, {})
|
|
model = profile_map.get(role) or profile_map.get("default")
|
|
if not model:
|
|
raise ValueError(f"no model mapped for profile={prof.value} role={role!r}")
|
|
return model
|
|
|
|
|
|
class AzureFoundryBackend:
|
|
"""AZURE profile: ``FoundryChatClient`` against a Foundry project (U18)."""
|
|
|
|
profile = Profile.AZURE
|
|
|
|
def create_chat_client(self, *, model: str) -> BaseChatClient:
|
|
endpoint = os.environ.get("PORTFOLIO_FOUNDRY_PROJECT_ENDPOINT")
|
|
if not endpoint:
|
|
raise ValueError(
|
|
"PORTFOLIO_FOUNDRY_PROJECT_ENDPOINT is required for the AZURE profile"
|
|
)
|
|
# Credential resolves lazily via Azure DefaultAzureCredential (az login / MI).
|
|
return FoundryChatClient(project_endpoint=endpoint, model=model)
|
|
|
|
|
|
class LocalBackend:
|
|
"""LOCAL profile: ``OpenAIChatCompletionClient`` against an OpenAI-compatible local
|
|
endpoint (Ollama/LM Studio). Development default per cost-discipline (D6)."""
|
|
|
|
profile = Profile.LOCAL
|
|
|
|
def create_chat_client(self, *, model: str) -> BaseChatClient:
|
|
base_url = os.environ.get("PORTFOLIO_LOCAL_BASE_URL", _DEFAULT_LOCAL_BASE_URL)
|
|
api_key = os.environ.get("PORTFOLIO_LOCAL_API_KEY", "ollama")
|
|
# Chat Completions (NOT the Responses-based OpenAIChatClient), non-streaming usage.
|
|
# Construction is offline — no network call until an agent actually runs.
|
|
return OpenAIChatCompletionClient(model=model, api_key=api_key, base_url=base_url)
|
|
|
|
|
|
def get_backend(profile: Profile | str) -> ChatBackend:
|
|
"""Select a backend by profile. Fail-fast (``ValueError``) on unknown profile."""
|
|
profile = Profile(profile) # validates: ValueError on unknown string
|
|
if profile is Profile.AZURE:
|
|
return AzureFoundryBackend()
|
|
return LocalBackend()
|