把 GPT-6 Astra、Sol 与 Luna 用于生产 Agent：Microsoft Foundry 的选型与路由方法

import json
import os
import sys
import time
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

ENDPOINT = os.environ['FOUNDRY_CHAT_URL']
API_KEY = os.environ['FOUNDRY_API_KEY']
AUTH_MODE = os.getenv('FOUNDRY_AUTH_MODE', 'api-key')

# 这些值应填写 Foundry 中真实的部署 ID。
# 根据自己的基准测试，让 Astra、Sol、Luna 对应到合适的路由。
MODELS = {
    'volume': os.environ['MODEL_VOLUME'],
    'complex': os.environ['MODEL_COMPLEX'],
    'review': os.environ['MODEL_REVIEW'],
}


def choose_route(text: str, expected_steps: int, risk: str) -> str:
    if risk == 'high':
        return 'review'
    if expected_steps >= 4 or len(text) >= 6000:
        return 'complex'
    return 'volume'


def call_model(model: str, prompt: str, attempts: int = 3) -> dict:
    payload = {
        'model': model,
        'messages': [
            {
                'role': 'system',
                'content': 'You are a production assistant. Be precise and do not invent tool results.'
            },
            {'role': 'user', 'content': prompt},
        ],
        'temperature': 0.2,
        'max_tokens': 800,
    }

    headers = {'Content-Type': 'application/json'}
    if AUTH_MODE == 'bearer':
        headers['Authorization'] = f'Bearer {API_KEY}'
    else:
        headers['api-key'] = API_KEY

    body = json.dumps(payload).encode('utf-8')

    for attempt in range(attempts):
        request = Request(ENDPOINT, data=body, headers=headers, method='POST')
        started = time.perf_counter()
        try:
            with urlopen(request, timeout=60) as response:
                result = json.loads(response.read().decode('utf-8'))
                result['_latency_ms'] = round(
                    (time.perf_counter() - started) * 1000, 1
                )
                return result
        except HTTPError as exc:
            retryable = exc.code == 429 or 500 <= exc.code < 600
            if not retryable or attempt == attempts - 1:
                detail = exc.read().decode('utf-8', errors='replace')
                raise RuntimeError(f'Foundry HTTP {exc.code}: {detail}') from exc
        except URLError as exc:
            if attempt == attempts - 1:
                raise RuntimeError(f'Foundry network error: {exc}') from exc

        time.sleep(2 ** attempt)

    raise RuntimeError('Request failed without a response')


def main() -> None:
    prompt = ' '.join(sys.argv[1:]).strip()
    if not prompt:
        raise SystemExit('Usage: python agent_router.py "your task"')

    steps = int(os.getenv('TASK_EXPECTED_STEPS', '1'))
    risk = os.getenv('TASK_RISK', 'low').lower()
    route = choose_route(prompt, steps, risk)
    model = MODELS[route]
    result = call_model(model, prompt)

    print(json.dumps({
        'route': route,
        'model': model,
        'latency_ms': result.pop('_latency_ms'),
        'response': result,
    }, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    main()

export FOUNDRY_CHAT_URL='https://YOUR-ENDPOINT.example/chat/completions'
export FOUNDRY_API_KEY='replace-with-your-key'
export FOUNDRY_AUTH_MODE='api-key'

export MODEL_VOLUME='your-benchmarked-deployment-id'
export MODEL_COMPLEX='your-benchmarked-deployment-id'
export MODEL_REVIEW='your-benchmarked-deployment-id'

TASK_EXPECTED_STEPS=2 TASK_RISK=low \
python agent_router.py '从这段客户反馈中提取产品名、问题类型和紧急程度，并返回 JSON。'

{
  "trace_id": "01J...",
  "workflow": "support-ticket-triage",
  "route": "complex",
  "model_deployment": "deployment-id",
  "latency_ms": 1840,
  "input_tokens": 2130,
  "output_tokens": 380,
  "tool_calls": 2,
  "retries": 0,
  "result": "success"
}