고객지원 AI를 평가하고 개선합니다 · v1

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 80–105 · GitHub에서 보기

def check_identity(config: Config, run: Callable = az_json) -> dict:
    account = run(["account", "show", "--subscription", config.subscription_id])
    if not isinstance(account, dict):
        raise LabError(text(
            "az account show 응답 형식이 올바르지 않습니다.",
            "The az account show response has an unexpected format.",
        ))
    actual_user = account.get("user", {}).get("name", "")
    checks = {
        "subscription": account.get("id") == config.subscription_id,
        "tenant": account.get("tenantId") == config.tenant_id,
        "user": actual_user.lower() == config.expected_user.lower(),
        "enabled": account.get("state") == "Enabled",
    }
    if not all(checks.values()):
        failed = ", ".join(key for key, ok in checks.items() if not ok)
        raise LabError(text(
            f"로그인 환경 불일치({failed}). 작업을 중단했습니다. "
            f"az login --tenant {config.tenant_id} 후 지정 계정으로 로그인해야 합니다. "
            "SDK는 다른 환경 자격 증명으로 자동 전환하지 않습니다.",
            f"The sign-in environment does not match ({failed}). Work stopped. "
            f"Run az login --tenant {config.tenant_id} and sign in with the designated account. "
            "The SDK never switches to another environment's credentials automatically.",
            language=config.language,
        ))
    return {"user": actual_user, "subscription": account["id"], "tenant": account["tenantId"]}

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 108–258 · GitHub에서 보기

def run_preflight(config: Config, *, run: Callable = az_json) -> dict:
    identity = check_identity(config, run)
    account = run([
        "cognitiveservices", "account", "show", "--subscription", config.subscription_id,
        "--resource-group", config.resource_group, "--name", config.account,
    ])
    project = run([
        "resource", "show", "--ids", config.project_id, "--api-version", "2025-06-01",
    ])
    search = run([
        "search", "service", "show", "--subscription", config.subscription_id,
        "--resource-group", config.resource_group, "--name", config.search_service,
    ])
    deployments = run([
        "cognitiveservices", "account", "deployment", "list",
        "--subscription", config.subscription_id, "--resource-group", config.resource_group,
        "--name", config.account,
    ])
    checks: list[dict] = []
    for name, resource in (("foundry", account), ("project", project), ("search", search)):
        region = str(resource.get("location", "")).lower().replace(" ", "")
        checks.append({
            "name": f"{name}_region",
            "status": "PASS" if region == config.location else "BLOCKED",
            "observed": region,
            "expected": config.location,
        })
    endpoint = project.get("properties", {}).get("endpoints", {}).get("AI Foundry API")
    checks.append({
        "name": "project_endpoint",
        "status": "PASS" if endpoint == config.project_endpoint else "BLOCKED",
        "observed": endpoint,
    })
    deployment_map = {d["name"]: d for d in deployments}
    model_evidence = {}
    roles = (
        ("agent", config.model), ("judge", config.judge), ("optimizer", config.optimizer),
        ("iq_planner", config.planner), ("embedding", config.embedding),
    )
    rate_limit_sources = {}
    for name in dict.fromkeys(name for _, name in roles):
        deployment = deployment_map.get(name)
        if not deployment:
            continue
        limits = deployment.get("properties", {}).get("rateLimits")
        source = "Azure CLI deployment metadata"
        if not isinstance(limits, list) or not any(
            isinstance(limit, dict) and limit.get("key") == "token" for limit in limits
        ):
            # Use raw ARM when CLI metadata lacks token/request labels instead of inferring TPM.
            identifier = f"{config.account_id}/deployments/{quote(name, safe='')}"
            deployment = run([
                "rest", "--method", "GET", "--url",
                f"https://management.azure.com{identifier}?api-version=2025-06-01",
            ])
            if (
                not isinstance(deployment, dict) or deployment.get("name") != name
                or str(deployment.get("id", "")).lower() != identifier.lower()
                or not isinstance(deployment.get("properties"), dict)
            ):
                raise LabError(text(
                    "TPM 조회 응답이 요청한 모델 배포와 다릅니다.",
                    "The TPM lookup response does not match the requested model deployment.",
                ))
            deployment_map[name] = deployment
            source = "Raw ARM deployment metadata"
        rate_limit_sources[name] = source
    for role, name in roles:
        deployment = deployment_map.get(name)
        ready = deployment and deployment.get("properties", {}).get("provisioningState") == "Succeeded"
        checks.append({
            "name": f"{role}_deployment", "status": "PASS" if ready else "BLOCKED",
            "observed": name,
        })
        if deployment:
            limits = deployment.get("properties", {}).get("rateLimits")
            minimum_tpm = LAB_MINIMUM_TPM[role]
            try:
                observed_tpm = _tokens_per_minute(limits)
                sufficient = observed_tpm >= minimum_tpm
                reason = (
                    "Meets the lab starting TPM requirement; RPM and shared load still apply."
                    if sufficient else
                    f"Allocate at least {minimum_tpm:,} TPM with operator approval, then rerun preflight."
                )
            except ValueError as exc:
                observed_tpm, sufficient, reason = None, False, str(exc)
            checks.append({
                "name": f"{role}_tpm", "status": "PASS" if sufficient else "BLOCKED",
                "observed": observed_tpm, "expected": minimum_tpm, "reason": reason,
            })
            model_evidence[role] = {
                "deployment": name,
                "model": deployment.get("properties", {}).get("model"),
                "sku": deployment.get("sku"),
                "tokens_per_minute": observed_tpm,
                "rate_limit_source": rate_limit_sources[name],
            }
        if role == "iq_planner":
            observed_model = (deployment or {}).get("properties", {}).get("model", {}).get("name")
            checks.append({
                "name": "iq_planner_supported_model",
                "status": "PASS" if observed_model == "gpt-5.5" else "BLOCKED",
                "observed": observed_model,
                "expected": "gpt-5.5",
            })
        if role == "embedding":
            observed_model = (deployment or {}).get("properties", {}).get("model", {}).get("name")
            checks.append({
                "name": "embedding_supported_model",
                "status": "PASS" if observed_model == "text-embedding-3-large" else "BLOCKED",
                "observed": observed_model, "expected": "text-embedding-3-large",
            })
    public_access = account.get("properties", {}).get("publicNetworkAccess")
    if public_access == "Disabled":
        checks.append({
            "name": "network", "status": "BLOCKED",
            "observed": text(
                "PublicNetworkAccess=Disabled: 승인된 VNet 연결 환경에서 실행해야 합니다.",
                "PublicNetworkAccess=Disabled: run from an approved VNet-connected environment.",
            ),
        })
    return {
        "checked_at": datetime.now(timezone.utc).isoformat(),
        "kind": "read-only-management-preflight",
        "status": "PASS" if all(c["status"] == "PASS" for c in checks) else "BLOCKED",
        "identity": identity,
        "checks": checks,
        "models": model_evidence,
        "identities": {
            "foundry_project_managed_identity": project.get("identity", {}).get("principalId"),
            "search_managed_identity": search.get("identity", {}).get("principalId"),
        },
        "not_verified": [
            text("데이터 평면 RBAC와 모델 추론 성공", "Data-plane RBAC and successful model inference"),
            text("Foundry IQ 지식 베이스 생성·검색 권한", "Foundry IQ knowledge-base creation and retrieval permissions"),
            text(
                "Prompt Optimizer 및 Frontier Tuning 테넌트 접근 권한",
                "Tenant access to Prompt Optimizer and Frontier Tuning",
            ),
            text(
                "Global Standard는 North Central US 내부 처리를 보장하지 않습니다.",
                "Global Standard does not guarantee processing inside North Central US.",
            ),
            text(
                "TPM 충족은 남은 토큰, 공유 부하, RPM·버스트 제한 또는 429 없는 실행을 보장하지 않습니다.",
                "Meeting the TPM minimum does not guarantee remaining tokens, absence of shared load, "
                "RPM or burst limits, or a run without 429 errors.",
            ),
        ],
    }

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 12–16 · GitHub에서 보기

def credential_for(config: Config) -> AzureCliCredential:
    check_identity(config)
    # Azure CLI rejects get-access-token when both tenant and subscription are
    # supplied. The identity check above binds the subscription to the tenant.
    return AzureCliCredential(subscription=config.subscription_id, process_timeout=30)

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 168–216 · GitHub에서 보기

def smoke_model(config: Config, run_id: str) -> dict:
    directory = safe_run_dir(run_id)
    path = directory / "model-smoke.json"
    if path.exists():
        previous = read_json(path)
        require_content_language(previous)
        if previous.get("project_endpoint") != config.project_endpoint or previous.get("deployment") != config.model:
            raise LabError(text(
                "기존 model smoke의 환경/배포가 다릅니다.",
                "The environment or deployment of the earlier model smoke check differs.",
            ))
        if previous.get("status") != "completed":
            raise LabError(text(
                "이 model smoke는 실패 또는 결과 불명입니다. 원본을 확인하기 전 재호출하지 않습니다.",
                "This model smoke check failed or its outcome is unknown. Do not call it again before checking the original record.",
            ))
        return previous
    snapshot = model_snapshot(config, config.model)
    record = {
        **language_metadata(),
        "kind": "LIVE_MODEL_SMOKE_NOT_QUALITY_EVALUATION", "status": "submitting",
        "started_at": datetime.now(timezone.utc).isoformat(),
        "code": code_provenance(),
        "project_endpoint": config.project_endpoint, "deployment": config.model,
        "model_snapshot": snapshot, "response_id": None, "cost": {"status": "NOT_OBSERVED"},
        "request": {"input": "Synthetic Contoso lab connectivity check. Reply with the single word READY.", "max_output_tokens": 128},
    }
    write_once_json(path, record)
    started = time.perf_counter()
    with credential_for(config) as credential:
        with AIProjectClient(endpoint=config.project_endpoint, credential=credential, retry_total=0) as project:
            with project.get_openai_client(max_retries=0, timeout=120.0) as client:
                response = client.responses.create(model=config.model, **record["request"], store=True)
                record["response_id"] = response.id
                record["response"] = response.model_dump(mode="json")
                record["usage"] = record["response"].get("usage")
                record["latency_ms"] = (time.perf_counter() - started) * 1000
                record["status"] = "completed" if (
                    record["response"].get("status") == "completed"
                    and isinstance(response.id, str) and response.id.strip()
                    and isinstance(response.output_text, str) and response.output_text.strip()
                ) else "failed"
                save_json(path, record)
    if record["status"] != "completed":
        raise LabError(text(
            "모델 smoke가 완료된 비어 있지 않은 응답을 반환하지 않았습니다.",
            "The model smoke check did not return a completed, non-empty response.",
        ))
    return record

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 33–108 · GitHub에서 보기

def knowledge_payloads(config: Config, names: dict, planner_model: str) -> dict:
    if not config.embedding:
        raise LabError(text(
            "벡터 검색을 위해 EMBEDDING_DEPLOYMENT를 명시해야 합니다.",
            "EMBEDDING_DEPLOYMENT must be set explicitly for vector search.",
        ))
    analyzer = f"{selected_language()}.microsoft"
    return {
        "index": {
            "name": names["index"],
            "fields": [
                {"name": "id", "type": "Edm.String", "key": True, "filterable": True, "retrievable": True},
                {"name": "title", "type": "Edm.String", "searchable": True, "retrievable": True, "analyzer": analyzer},
                {"name": "content", "type": "Edm.String", "searchable": True, "retrievable": True, "analyzer": analyzer},
                {"name": "effective_date", "type": "Edm.String", "filterable": True, "retrievable": True},
                {"name": "content_vector", "type": "Collection(Edm.Single)", "searchable": True,
                 "retrievable": False, "dimensions": DIMENSIONS, "vectorSearchProfile": "contoso-vector"},
            ],
            "vectorSearch": {
                "algorithms": [{"name": "contoso-hnsw", "kind": "hnsw", "hnswParameters": {"metric": "cosine"}}],
                "profiles": [{"name": "contoso-vector", "algorithm": "contoso-hnsw", "vectorizer": "contoso-embedding"}],
                "vectorizers": [{
                    "name": "contoso-embedding", "kind": "azureOpenAI",
                    "azureOpenAIParameters": {
                        "resourceUri": config.openai_endpoint, "deploymentId": config.embedding,
                        "modelName": "text-embedding-3-large",
                    },
                }],
            },
            "semantic": {
                "defaultConfiguration": "support-semantic",
                "configurations": [{
                    "name": "support-semantic",
                    "prioritizedFields": {
                        "titleField": {"fieldName": "title"},
                        "prioritizedContentFields": [{"fieldName": "content"}],
                    },
                }],
            },
        },
        "source": {
            "name": names["source"],
            "kind": "searchIndex",
            "searchIndexParameters": {
                "searchIndexName": names["index"],
                "semanticConfigurationName": "support-semantic",
                "sourceDataFields": [{"name": field} for field in ("id", "title", "content", "effective_date")],
                "searchFields": [{"name": field} for field in ("title", "content", "content_vector")],
            },
        },
        "base": {
            "name": names["base"],
            "description": "Synthetic Contoso policies. Stable ATLAS document IDs are in titles and source data.",
            "knowledgeSources": [{"name": names["source"]}],
            "models": [{
                "kind": "azureOpenAI",
                "azureOpenAIParameters": {
                    "resourceUri": config.openai_endpoint,
                    "deploymentId": config.planner,
                    "modelName": planner_model,
                },
            }],
            "retrievalReasoningEffort": {"kind": "low"},
            "outputMode": "extractiveData",
        },
        "connection": {
            "properties": {
                "authType": "ProjectManagedIdentity",
                "category": "RemoteTool",
                "target": search_url(config, "knowledgebases", names["base"], "/mcp"),
                "isSharedToAll": False,
                "audience": "https://search.azure.com/",
                "metadata": {"ApiType": "Azure"},
            },
        },
    }

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 111–134 · GitHub에서 보기

def _ensure_created(config: Config, http: JsonHttp, kind: str, name: str, url: str, payload: dict) -> dict:
    state = workspace(config)
    owned = [item for item in state["created"] if item.get("kind") == kind and item.get("name") == name]
    try:
        existing = http.request("GET", url)
    except CloudRequestError as exc:
        if exc.status != 404:
            raise
        existing = None
    if existing is not None:
        if not owned:
            raise LabError(text(
                f"{name}이 이미 존재하지만 이 실습의 생성 기록이 없습니다. 덮어쓰지 않습니다.",
                f"{name} already exists, but this lab has no creation record for it. Not overwriting.",
            ))
        return existing.body
    result = http.request("PUT", url, payload, create_only=True)
    if result.status not in {200, 201}:
        raise LabError(text(
            f"{name}: 비동기/예상 밖 상태 {result.status}. 완료로 기록하지 않습니다.",
            f"{name}: asynchronous or unexpected status {result.status}. Not recorded as complete.",
        ))
    record_created(config, {"kind": kind, "name": name, "url": url, "etag": result.etag})
    return result.body

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 137–274 · GitHub에서 보기

def prepare_knowledge(config: Config) -> dict:
    state = workspace(config, create=True)
    names = resource_names(config, state)
    documents_path = content_path(ROOT, "data/knowledge/documents.json")
    documents = read_json(documents_path)
    content_hash = sha256_file(documents_path)
    setup_path = artifacts_dir() / "knowledge/setup.json"
    previous = None
    if setup_path.exists():
        previous = read_json(setup_path)
        require_content_language(previous)
        if previous.get("documents_sha256") != content_hash or previous.get("names") != names:
            raise LabError(text(
                "기존 지식 준비와 원본/범위가 다릅니다. 다른 실험을 기존 인덱스에 덮어쓰지 않습니다.",
                "The source or scope differs from the earlier knowledge preparation. A different experiment is never written over the existing index.",
            ))
        if previous.get("status") in {"created_not_retrieval_tested", "retrieval_verified"}:
            retrieval = previous.get("retrieval", {})
            if (
                retrieval.get("embedding_deployment") != config.embedding
                or retrieval.get("embedding_model") != "text-embedding-3-large"
                or retrieval.get("dimensions") != DIMENSIONS
            ):
                raise LabError(text(
                    "준비된 인덱스의 embedding 계약이 다릅니다. 새 실험을 생성해야 합니다.",
                    "The embedding contract of the prepared index differs. Create a new experiment.",
                ))
            return previous
    with credential_for(config) as credential:
        planner = az_json([
            "cognitiveservices", "account", "deployment", "show",
            "--subscription", config.subscription_id, "--resource-group", config.resource_group,
            "--name", config.account, "--deployment-name", config.planner,
        ])
        model = planner.get("properties", {}).get("model", {})
        if not model.get("name"):
            raise LabError(text(
                "IQ planner의 실제 모델 이름을 확인할 수 없습니다.",
                "Could not confirm the actual model name of the IQ planner.",
            ))
        # This workshop's main lane was researched for gpt-5.5, not arbitrary
        # answer models that may be unsupported by the knowledge-base planner.
        if model["name"] != "gpt-5.5":
            raise LabError(text(
                "이 경로의 IQ planner는 확인된 gpt-5.5 배포를 요구합니다. 다른 모델로 자동 전환하지 않습니다.",
                "The IQ planner on this path requires a verified gpt-5.5 deployment. It never switches to another model automatically.",
            ))
        payloads = knowledge_payloads(config, names, model["name"])
        payload_hash = hashlib.sha256(json.dumps(payloads, sort_keys=True).encode()).hexdigest()
        if previous and previous.get("payload_sha256") != payload_hash:
            raise LabError(text(
                "미완료 검색 설정의 구성 계약이 바뀌었습니다. 같은 인덱스에 덮어쓰지 않습니다.",
                "The configuration contract of the unfinished search setup changed. Not overwriting the same index.",
            ))
        setup = {
            **language_metadata(),
            "status": "preparing", "workspace_id": state["workspace_id"], "names": names,
            "project_id": config.project_id, "documents_sha256": content_hash,
            "planner": {"deployment": config.planner, "model": model},
            "search_api_version": SEARCH_API,
            "payload_sha256": payload_hash,
            "retrieval": {
                "reasoning": "low", "output": "extractiveData",
                "embedding_model": "text-embedding-3-large", "embedding_deployment": config.embedding,
                "dimensions": DIMENSIONS, "index_algorithm": "hnsw", "metric": "cosine",
                "query_modes": ["vector", "hybrid", "agentic-iq-mcp"],
                "direct_probe_is_agent_evidence": False,
            },
        }
        save_json(setup_path, setup)
        save_json(artifacts_dir() / "knowledge/config-snapshot.json", payloads)
        search = JsonHttp(credential, scope=SEARCH_SCOPE, allowed_origin=config.search_endpoint)
        index_url = search_url(config, "indexes", names["index"])
        _ensure_created(config, search, "search_index", names["index"], index_url, payloads["index"])
        embedding = embed(
            config, [f"{doc['title']}\n{doc['content']}" for doc in documents],
            artifacts_dir() / "knowledge/document-embeddings.json",
        )
        vectors = validate_vectors(embedding["response"], len(documents))
        uploaded = [{
            "@search.action": "mergeOrUpload",
            "id": doc["id"], "title": f"{doc['id']} | {doc['title']}",
            "content": f"[{doc['id']}]\n{doc['content']}",
            "effective_date": doc["effective_date"],
            "content_vector": vector,
        } for doc, vector in zip(documents, vectors, strict=True)]
        upload_path = artifacts_dir() / "knowledge/upload-response.json"
        upload_contract = {"documents_sha256": content_hash, "payload_sha256": payload_hash,
                           "embedding_sha256": sha256_file(artifacts_dir() / "knowledge/document-embeddings.json")}
        if upload_path.exists():
            upload_record = read_json(upload_path)
            if upload_record.get("contract") != upload_contract or upload_record.get("status") != "completed":
                raise LabError(text(
                    "업로드 계약 변경 또는 결과 불명입니다. 업로드를 반복하지 않고 실제 인덱스 상태를 확인해야 합니다.",
                    "The upload contract changed or its outcome is unknown. Do not repeat the upload; check the actual index state.",
                ))
            upload_body = upload_record["response"]
        else:
            write_once_json(upload_path, {"status": "submitting", "contract": upload_contract})
            upload = search.request(
                "POST", search_url(config, "indexes", names["index"], "/docs/index"), {"value": uploaded},
            )
            upload_body = upload.body
            save_json(upload_path, {"status": "response_received", "contract": upload_contract, "response": upload_body})
        statuses = upload_body.get("value", [])
        if (
            len(statuses) != len(uploaded)
            or {item.get("key") for item in statuses} != {doc["id"] for doc in documents}
            or any(item.get("status") is not True for item in statuses)
        ):
            save_json(artifacts_dir() / "knowledge/upload-error.json", upload_body)
            raise LabError(text(
                "일부 문서 업로드가 실패했습니다. knowledge/upload-error.json을 확인해야 합니다.",
                "Some document uploads failed. Check knowledge/upload-error.json.",
            ))
        save_json(upload_path, {"status": "completed", "contract": upload_contract, "response": upload_body})
        _ensure_created(
            config, search, "knowledge_source", names["source"],
            search_url(config, "knowledgesources", names["source"]), payloads["source"],
        )
        _ensure_created(
            config, search, "knowledge_base", names["base"],
            search_url(config, "knowledgebases", names["base"]), payloads["base"],
        )
        connection_url = (
            f"https://management.azure.com{config.project_id}/connections/"
            f"{names['connection']}?api-version={CONNECTION_API}"
        )
        arm = JsonHttp(credential, scope=ARM_SCOPE, allowed_origin="https://management.azure.com")
        _ensure_created(config, arm, "project_connection", names["connection"], connection_url, payloads["connection"])
        setup["status"] = "created_not_retrieval_tested"
        setup["mcp_endpoint"] = payloads["connection"]["properties"]["target"]
        setup["connection_name"] = names["connection"]
        setup["uploaded_documents"] = len(uploaded)
        setup["embedding_usage"] = embedding["response"].get("usage")
        save_json(setup_path, setup)
        save_json(artifacts_dir() / "knowledge/documents-snapshot.json", documents)
        return setup

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 51–122 · GitHub에서 보기

def embed(config: Config, texts: list[str], cache_path: Path) -> dict:
    if not config.embedding or not texts or any(not isinstance(text, str) or not text.strip() for text in texts):
        raise LabError(localized_text(
            "EMBEDDING_DEPLOYMENT와 비어 있지 않은 embedding 입력이 필요합니다.",
            "EMBEDDING_DEPLOYMENT and non-empty embedding input are required.",
        ))
    deployment = model_snapshot(config, config.embedding)
    if deployment["model"].get("name") != "text-embedding-3-large":
        raise LabError(localized_text(
            "이 인덱스의 동결 계약은 text-embedding-3-large/3072입니다. 자동 대체하지 않습니다.",
            "This index's frozen contract is text-embedding-3-large/3072. It is never replaced automatically.",
        ))
    contract = {
        "endpoint": config.openai_endpoint, "deployment": deployment,
        "dimensions": DIMENSIONS,
        "input_sha256": hashlib.sha256(json.dumps(texts, ensure_ascii=False).encode()).hexdigest(),
        "count": len(texts),
    }
    if cache_path.exists():
        previous = read_json(cache_path)
        if previous.get("contract") != contract:
            raise LabError(localized_text(
                "기존 embedding 실행과 입력/모델/범위가 다릅니다. 기존 기록을 보존해야 합니다.",
                "The input, model or scope differs from the earlier embedding run. Preserve the existing record.",
            ))
        if previous.get("status") != "completed":
            raise LabError(localized_text(
                "이 embedding 요청은 실패 또는 결과 불명입니다. 원격 요청을 확인하기 전 재호출하지 않습니다.",
                "This embedding request failed or its outcome is unknown. Do not call it again before checking the remote request.",
            ))
        validate_vectors(previous["response"], len(texts))
        return previous
    from azure.identity import get_bearer_token_provider
    from openai import APIError, OpenAI
    from azure.core.exceptions import AzureError
    from lab.auth import credential_for

    record = {"kind": "LIVE_AZURE_EMBEDDING", "contract": contract, "status": "submitting",
              "code": code_provenance(),
              "started_at": datetime.now(timezone.utc).isoformat(), "cost": {"status": "NOT_OBSERVED"}}
    cache_path.parent.mkdir(parents=True, exist_ok=True)
    try:
        with cache_path.open("x", encoding="utf-8") as stream:
            json.dump(record, stream, ensure_ascii=False, indent=2)
    except FileExistsError as exc:
        raise LabError(localized_text(
            "동시 embedding 실행을 차단했습니다. 기존 기록을 확인해야 합니다.",
            "A concurrent embedding run was blocked. Check the existing record.",
        )) from exc
    started = time.perf_counter()
    try:
        with credential_for(config) as credential:
            with OpenAI(
                base_url=f"{config.openai_endpoint}/openai/v1/",
                api_key=get_bearer_token_provider(credential, "https://ai.azure.com/.default"),
                max_retries=0, timeout=120,
            ) as client:
                response = client.embeddings.create(
                    model=config.embedding, input=texts, dimensions=DIMENSIONS, encoding_format="float",
                )
                record["response"] = response.model_dump(mode="json")
                record["request_id"] = getattr(response, "_request_id", None)
                save_json(cache_path, record)
                validate_vectors(record["response"], len(texts))
    except (APIError, AzureError, LabError) as exc:
        record.update(status="failed_or_unknown", error={"type": type(exc).__name__, "message": str(exc)})
        save_json(cache_path, record)
        raise
    record.update(status="completed", latency_ms=(time.perf_counter() - started) * 1000,
                  completed_at=datetime.now(timezone.utc).isoformat())
    save_json(cache_path, record)
    return record

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 400–414 · GitHub에서 보기

def knowledge_tool(config: Config) -> MCPTool:
    setup = load_knowledge(config)
    if setup["status"] != "retrieval_verified":
        raise LabError(text(
            "먼저 python -m lab iq probe --confirm 으로 실제 검색을 확인해야 합니다.",
            "First verify real retrieval with python -m lab iq probe --confirm.",
        ))
    return MCPTool(
        server_label="contoso-knowledge",
        server_url=setup["mcp_endpoint"],
        server_description="Read-only Contoso Atlas Cloud policies. Cite stable ATLAS document IDs shown in the returned title/content, not numeric retrieval reference IDs.",
        allowed_tools=["knowledge_base_retrieve"],
        require_approval="never",
        project_connection_id=setup["connection_name"],
    )

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 124–165 · GitHub에서 보기

def create_native_agent(config: Config, version: str, prompt: Path) -> dict:
    """Connect the prepared policy tool to a fixed, strict-output workshop Agent."""
    from lab.knowledge import knowledge_tool

    if version not in {"1", "2"}:
        raise LabError("The native workshop Agent permits only versions 1 and 2.")
    if not prompt.is_file() or not prompt.read_text(encoding="utf-8").strip():
        raise LabError(f"A nonempty instruction file is required: {prompt}")
    state = workspace(config)
    tool = knowledge_tool(config)
    snapshot = model_snapshot(config, config.model)
    snapshot_path = artifacts_dir() / "agents/native-model.json"
    if snapshot_path.exists():
        if read_json(snapshot_path) != snapshot:
            raise LabError("The Agent model deployment changed after setup; preserve the original comparison.")
    elif version == "2":
        raise LabError("Create and verify native Agent v1 before preparing v2.")
    else:
        write_once_json(snapshot_path, snapshot)
    definition = {
        "kind": "prompt",
        "model": config.model,
        "instructions": prompt.read_text(encoding="utf-8"),
        "tools": [tool.as_dict()],
        "text": native_response_format(),
    }
    receipt = artifacts_dir() / "agents" / f"native-v{version}.json"
    with credential_for(config) as credential:
        with AIProjectClient(endpoint=config.project_endpoint, credential=credential, retry_total=0) as project:
            result = ensure_fixed_release(
                project, project_endpoint=config.project_endpoint, agent_name=config.agent_name("iq"),
                version=version, definition=definition, receipt=receipt,
                workspace_id=state["workspace_id"],
            )
    created = {"kind": "agent_version", "name": config.agent_name("iq"), "version": version}
    if created not in workspace(config)["created"]:
        record_created(config, created)
    return {
        "status": result["status"], "agent_name": config.agent_name("iq"), "version": version,
        "receipt": str(receipt), "model_snapshot": snapshot,
        "evaluation_status": "NOT_RUN", "production_approval": "NOT_GRANTED",
    }

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 49–121 · GitHub에서 보기

def ensure_fixed_release(project: AIProjectClient, *, project_endpoint: str, agent_name: str,
                         version: str, definition: dict, receipt: Path,
                         workspace_id: str | None = None) -> dict:
    if version not in {"1", "2"}:
        raise LabError("This workshop permits only released v1 and v2; use drafts for experiments.")
    if definition.get("kind") != "prompt" or not isinstance(definition.get("instructions"), str) or not definition["instructions"].strip():
        raise LabError("A nonempty prompt-agent definition is required.")
    expected_definition = _definition_for_comparison(definition)
    intent = {
        "project_endpoint": project_endpoint, "agent_name": agent_name, "version": version,
        "definition_sha256": hashlib.sha256(
            json.dumps(definition, sort_keys=True, separators=(",", ":")).encode()
        ).hexdigest(),
    }
    if workspace_id is not None:
        intent["workspace_id"] = workspace_id
    previous = read_json(receipt) if receipt.exists() else None
    if previous is not None and previous["intent"] != intent:
        raise LabError("The fixed-version receipt belongs to a different definition or environment.")
    try:
        existing = project.agents.get_version(agent_name=agent_name, agent_version=version).as_dict()
    except ResourceNotFoundError:
        existing = None
    if existing is not None:
        if workspace_id is not None and (existing.get("metadata") or {}).get("workspace") != workspace_id:
            raise LabError("The existing Agent version belongs to a different workspace; no changes were made.")
        if _definition_for_comparison(existing["definition"]) != expected_definition:
            raise LabError("The released version is immutable and differs from this source; no new version was created.")
        record = {"intent": intent, "status": "verified", "agent": existing}
        if previous is None:
            write_once_json(receipt, record)
        else:
            save_json(receipt, record)
        return record
    if previous is not None:
        raise LabError("The creation outcome is unknown; inspect the receipt and remote state, do not resubmit.")
    try:
        releases = [
            item.as_dict() for item in project.agents.list_versions(agent_name=agent_name)
            if not item.draft and not item.version.startswith("draft-")
        ]
    except ResourceNotFoundError:
        releases = []
    if {item["version"] for item in releases} != ({"1"} if version == "2" else set()):
        raise LabError("Unexpected release history; refusing to create v3 or reset existing versions.")
    if version == "2":
        if workspace_id is not None and (releases[0].get("metadata") or {}).get("workspace") != workspace_id:
            raise LabError("The baseline Agent belongs to a different workspace; no changes were made.")
        baseline = _definition_for_comparison(releases[0]["definition"])
        if {key: value for key, value in baseline.items() if key != "instructions"} != {
            key: value for key, value in expected_definition.items() if key != "instructions"
        }:
            raise LabError("V1 and v2 must keep the same model, tools and generation settings.")
        if baseline["instructions"] == definition["instructions"]:
            raise LabError("Identical instructions are not a new improvement candidate.")
    record = {"intent": intent, "status": "creating", "agent": None}
    write_once_json(receipt, record)
    metadata = {"lab": "foundry-evaluation-optimizer", "release_policy": "fixed-v1-v2"}
    if workspace_id is not None:
        metadata["workspace"] = workspace_id
    created = project.agents.create_version(
        agent_name=agent_name, definition=PromptAgentDefinition(definition), draft=False,
        metadata=metadata,
        description=f"Fixed workshop v{version}; separate managed evaluation is required.",
    ).as_dict()
    record.update(status="created", agent=created)
    save_json(receipt, record)
    if (
        created["version"] != version or _definition_for_comparison(created["definition"]) != expected_definition
        or (workspace_id is not None and (created.get("metadata") or {}).get("workspace") != workspace_id)
    ):
        raise LabError("The created version differs from the requested fixed version; receipt preserved.")
    return record

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 42–83 · GitHub에서 보기

def evaluation_contract(evaluation: dict) -> dict:
    expected = {
        "Relevance": ("builtin.relevance", 4, "{{sample.output_text}}"),
        "TaskAdherence": ("builtin.task_adherence", 1, "{{sample.output_items}}"),
    }
    criteria = evaluation.get("testing_criteria")
    if not isinstance(criteria, list) or len(criteria) != len(expected):
        raise ValueError("Use exactly the workshop's Relevance and TaskAdherence evaluators.")
    judges = set()
    seen = set()
    for criterion in criteria:
        if not isinstance(criterion, dict):
            raise ValueError("Each Foundry evaluation criterion must be a JSON object.")
        name = criterion.get("name")
        if not isinstance(name, str) or name not in expected or name in seen:
            raise ValueError("Evaluator names must be unique Relevance and TaskAdherence entries.")
        seen.add(name)
        evaluator, threshold, response = expected[name]
        parameters = criterion.get("initialization_parameters") or {}
        mapping = criterion.get("data_mapping") or {}
        if (
            criterion.get("type") != "azure_ai_evaluator"
            or criterion.get("evaluator_name") != evaluator
            or not isinstance(parameters, dict) or not isinstance(mapping, dict)
            or type(parameters.get("threshold")) not in (int, float)
            or parameters["threshold"] != threshold
            or mapping.get("query") != "{{item.query}}"
            or mapping.get("response") != response
        ):
            raise ValueError(f"{name} has a different evaluator, threshold, or response mapping.")
        judge = parameters.get("deployment_name")
        if not isinstance(judge, str) or not judge.strip():
            raise ValueError(f"{name} has no explicit Judge deployment in the remote definition.")
        judges.add(judge)
    if len(judges) != 1:
        raise ValueError("Both workshop evaluators must use the same explicit Judge deployment.")
    return {
        "judge_deployment": judges.pop(),
        "thresholds": {"Relevance": 4, "TaskAdherence": 1},
        "source": "existing_remote_evaluation_definition_not_local_environment_defaults",
        "testing_criteria_sha256": digest({"criteria": criteria}),
    }

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 89–112 · GitHub에서 보기

def list_native_evaluations(config: Config, name: str) -> dict:
    """Read portal-created evaluation identities without submitting or saving runs."""
    if not name.strip():
        raise LabError("Specify the exact evaluation name shown in the Foundry portal.")
    evaluations = []
    with credential_for(config) as credential:
        with AIProjectClient(endpoint=config.project_endpoint, credential=credential, retry_total=0) as project:
            with project.get_openai_client(max_retries=0, timeout=60) as client:
                for evaluation in client.evals.list(limit=100):
                    if evaluation.name != name:
                        continue
                    runs = []
                    for run in client.evals.runs.list(eval_id=evaluation.id, limit=100):
                        value = run.model_dump(mode="json", warnings=False)
                        target = (value.get("data_source") or {}).get("target") or {}
                        runs.append({
                            "run_id": run.id, "name": run.name, "status": run.status,
                            "agent_name": target.get("name"), "agent_version": target.get("version"),
                            "result_counts": value.get("result_counts"),
                        })
                    evaluations.append({"evaluation_id": evaluation.id, "name": evaluation.name, "runs": runs})
    if not evaluations:
        raise LabError("No evaluation with that exact name exists in this project; check the portal name and project.")
    return {"mode": "READ_ONLY", "evaluations": evaluations}

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 21–34 · GitHub에서 보기

def native_response_format() -> dict:
    contract = read_json(ROOT / "schemas/response.schema.json")
    schema = {
        key: deepcopy(contract[key])
        for key in ("type", "additionalProperties", "required", "properties")
    }
    # Routing conditions and citation uniqueness remain validation requirements;
    # the model's structured-output subset does not support allOf/uniqueItems.
    schema["properties"]["citations"].pop("uniqueItems")
    schema["properties"]["route"]["type"] = "string"
    return {"format": {
        "type": "json_schema", "name": "contoso_support_response", "strict": True,
        "schema": schema,
    }}

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 1–26 · GitHub에서 보기

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "$id": "urn:foundry-workshop:response:v1",
  "title": "Workshop agent response",
  "description": "Exactly one unfenced JSON object. Citation IDs are lexical references, not proof of semantic groundedness. Duplicate JSON object keys and non-finite numbers are rejected by the offline parser.",
  "type": "object",
  "additionalProperties": false,
  "required": ["answer", "citations", "route", "needs_human"],
  "properties": {
    "answer": {"type": "string"},
    "citations": {
      "type": "array",
      "uniqueItems": true,
      "items": {"type": "string", "minLength": 1, "pattern": "\\S"}
    },
    "route": {"enum": ["answer", "clarify", "escalate", "refuse"]},
    "needs_human": {"type": "boolean"}
  },
  "allOf": [
    {
      "if": {"properties": {"route": {"const": "escalate"}}, "required": ["route"]},
      "then": {"properties": {"needs_human": {"const": true}}},
      "else": {"properties": {"needs_human": {"const": false}}}
    }
  ]
}

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 111–138 · GitHub에서 보기

def candidate_source(baseline: dict, *, evaluation_id: str, version: str) -> dict:
    if baseline.get("eval_id") != evaluation_id or baseline.get("status") != "completed":
        raise ValueError("Use a completed baseline run from this exact Foundry evaluation.")
    source = deepcopy(baseline.get("data_source"))
    if not isinstance(source, dict) or source.get("type") != "azure_ai_target_completions":
        raise ValueError("The baseline must be an agent-targeted Foundry evaluation.")
    target = source.get("target", {})
    if target.get("type") != "azure_ai_agent" or not target.get("name"):
        raise ValueError("The baseline must identify a Foundry agent.")
    original = str(target.get("version", ""))
    pinned_version = r"(?:[1-9]\d*|draft-[1-9]\d*)"
    if not re.fullmatch(pinned_version, original) or not re.fullmatch(pinned_version, version):
        raise ValueError("Pin explicit numeric or draft agent versions, not latest.")
    if original == version:
        raise ValueError("The candidate version must differ from the baseline.")
    dataset = source.get("source", {})
    if dataset.get("type") != "file_id" or not dataset.get("id"):
        raise ValueError("The baseline must use an identified, registered dataset.")
    messages = source.get("input_messages", {})
    template = messages.get("template", [])
    if messages.get("type") != "template" or len(template) != 1 or template[0].get("role") != "user":
        raise ValueError("Use one query-only user message, without reference-answer leakage.")
    content = template[0].get("content")
    query = content if isinstance(content, str) else content.get("text") if isinstance(content, dict) else None
    if query != "{{item.query}}":
        raise ValueError("This workshop sends only {{item.query}} to the agent.")
    target["version"] = version
    return source

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 141–193 · GitHub에서 보기

def submit_or_resume(project, client, *, evaluation_id: str, baseline_run_id: str,
                     version: str, name: str, output: Path) -> dict:
    intent = {
        "evaluation_id": evaluation_id, "baseline_run_id": baseline_run_id,
        "candidate_version": version, "name": name,
    }
    if output.exists():
        saved = json.loads(output.read_text(encoding="utf-8"))
        if saved.get("intent") != intent:
            raise ValueError("This receipt belongs to a different evaluation or candidate.")
        if not saved.get("run_id"):
            raise ValueError("Submission outcome is unknown. Inspect Foundry before any new submission; preserve this receipt.")
        return saved
    baseline = as_object(client.evals.runs.retrieve(run_id=baseline_run_id, eval_id=evaluation_id))
    source = candidate_source(baseline, evaluation_id=evaluation_id, version=version)
    agent_name = source["target"]["name"]
    original_version = str(baseline["data_source"]["target"]["version"])
    before = as_object(project.agents.get_version(agent_name=agent_name, agent_version=original_version))["definition"]
    after = as_object(project.agents.get_version(agent_name=agent_name, agent_version=version))["definition"]
    if not isinstance(after.get("instructions"), str) or not after["instructions"].strip():
        raise ValueError("The candidate has no actual instructions.")
    if {key: value for key, value in before.items() if key != "instructions"} != {
        key: value for key, value in after.items() if key != "instructions"
    }:
        raise ValueError("Model, tools, or other agent settings changed. This is not an instruction-only comparison.")
    for existing in client.evals.runs.list(eval_id=evaluation_id, limit=100):
        existing = as_object(existing)
        if existing.get("name") == name:
            raise ValueError(
                f"Foundry already has a run named {name}: {existing.get('id')}. "
                "Open that run or restore its original receipt; do not create a duplicate."
            )
    evaluation = as_object(client.evals.retrieve(evaluation_id))
    contract = evaluation_contract(evaluation)
    saved = {
        "intent": intent,
        "status": "submitting",
        "run_id": None,
        "agent_name": agent_name,
        "dataset": source["source"],
        "testing_criteria_sha256": contract["testing_criteria_sha256"],
        "evaluation_contract": contract,
        "request": {"eval_id": evaluation_id, "name": name, "data_source": source},
        "production_approval": "NOT_GRANTED",
    }
    write_once_json(output, saved)
    # An interrupted POST intentionally leaves an unknown outcome, not permission to retry.
    run = as_object(client.evals.runs.create(**saved["request"]))
    if not run.get("id") or run.get("eval_id") != evaluation_id:
        raise ValueError("Foundry returned no matching evaluation-run identity; preserve the submission receipt.")
    saved.update(status=run["status"], run_id=run["id"], run=run)
    save_json(output, saved)
    return saved

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 215–295 · GitHub에서 보기

def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--config", type=Path,
        help="Environment .env from bootstrap; supplies the project endpoint, subscription and receipt folder",
    )
    parser.add_argument("--endpoint", help="The same Foundry project endpoint as the baseline (not needed with --config)")
    parser.add_argument("--subscription", help="The explicitly selected Azure CLI subscription (not needed with --config)")
    parser.add_argument("--evaluation", required=True, help="Existing eval_... ID; no new evaluation definition is created")
    parser.add_argument("--baseline", required=True, help="Completed baseline evalrun_... ID")
    parser.add_argument("--version", required=True, help="Explicit candidate agent version")
    parser.add_argument("--name", default="candidate-v2")
    parser.add_argument("--out", type=Path, help="Receipt path; defaults to the environment's foundry-evaluations folder")
    parser.add_argument("--wait-seconds", type=int, default=1800)
    args = parser.parse_args(argv)
    try:
        args.endpoint, args.subscription, args.out = settings(args)
        endpoint = urlsplit(args.endpoint)
        if (
            endpoint.scheme != "https" or endpoint.username or endpoint.password
            or endpoint.port or endpoint.query or endpoint.fragment
            or not re.fullmatch(r"[a-z0-9-]+\.services\.ai\.azure\.com", endpoint.hostname or "")
            or not re.fullmatch(r"/api/projects/[A-Za-z0-9_.()-]+/?", endpoint.path)
        ):
            raise ValueError("Use the exact HTTPS Foundry project endpoint, without credentials or query parameters.")
        if not 1 <= args.wait_seconds <= 3600:
            raise ValueError("--wait-seconds must be between 1 and 3600.")
        if args.config is None:
            account = azcli.run(["account", "show", "--subscription", args.subscription, "-o", "json"], timeout=30)
            if account.returncode:
                raise ValueError(f"Azure CLI could not read the subscription: {account.stderr.strip()}")
            identity = json.loads(account.stdout)
            if identity.get("id", "").lower() != args.subscription.lower() or identity.get("state") != "Enabled":
                raise ValueError("Azure CLI did not confirm the intended enabled subscription.")
        from azure.ai.projects import AIProjectClient
        from azure.identity import AzureCliCredential

        with AzureCliCredential(subscription=args.subscription, process_timeout=30) as credential:
            with AIProjectClient(endpoint=args.endpoint, credential=credential, retry_total=0) as project:
                with project.get_openai_client(max_retries=0, timeout=120) as client:
                    saved = submit_or_resume(
                        project, client, evaluation_id=args.evaluation, baseline_run_id=args.baseline,
                        version=args.version, name=args.name, output=args.out,
                    )
                    print(f"Foundry evaluation: {args.evaluation}\nRun: {saved['run_id']}", flush=True)
                    deadline = time.monotonic() + args.wait_seconds
                    while True:
                        run = as_object(client.evals.runs.retrieve(run_id=saved["run_id"], eval_id=args.evaluation))
                        saved.update(status=run["status"], run=run)
                        save_json(args.out, saved)
                        if run["status"] in {"completed", "failed", "canceled", "cancelled"}:
                            if run["status"] == "completed":
                                items = [
                                    as_object(item) for item in client.evals.runs.output_items.list(
                                        run_id=saved["run_id"], eval_id=args.evaluation,
                                    )
                                ]
                                saved["output_items"] = items
                                save_json(args.out, saved)
                                if len(items) != (run.get("result_counts") or {}).get("total"):
                                    raise ValueError("Completed-run output is incomplete; preserve the receipt.")
                                agent = as_object(project.agents.get_version(
                                    agent_name=saved["agent_name"], agent_version=args.version,
                                ))
                                saved["execution_verification"] = verify_agent_execution(
                                    items, agent_name=saved["agent_name"], version=args.version,
                                    instructions=agent["definition"]["instructions"],
                                )
                                save_json(args.out, saved)
                            print(json.dumps({
                                "status": run["status"], "result_counts": run.get("result_counts"),
                                "report_url": run.get("report_url"), "production_approval": "NOT_GRANTED",
                            }, indent=2))
                            return 0 if run["status"] == "completed" else 1
                        if time.monotonic() >= deadline:
                            print("Still running. Repeat this exact command to collect the same run; do not delete the receipt.", file=sys.stderr)
                            return 2
                        time.sleep(15)
    except (OSError, ValueError, subprocess.SubprocessError, LabError) as error:
        print(f"ERROR: {error}", file=sys.stderr)
        return 1

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 18–90 · GitHub에서 보기

def cleanup_plan(config: Config) -> dict:
    state = workspace(config)
    names = resource_names(config, state)
    allowed = {
        "search_index": (names["index"], search_url(config, "indexes", names["index"])),
        "knowledge_source": (names["source"], search_url(config, "knowledgesources", names["source"])),
        "knowledge_base": (names["base"], search_url(config, "knowledgebases", names["base"])),
        "project_connection": (
            names["connection"],
            f"https://management.azure.com{config.project_id}/connections/{names['connection']}?api-version={CONNECTION_API}",
        ),
    }
    allowed_agents = {config.agent_name(stage) for stage in ("baseline", "iq", "optimized", "tuned")}
    actions = []
    for record in state["created"]:
        kind = record.get("kind")
        if kind == "agent_version":
            if record.get("name") not in allowed_agents or not re.fullmatch(r"[A-Za-z0-9_-]+", str(record.get("version", ""))):
                raise LabError(text(
                    "정리 기록의 에이전트 범위가 다릅니다. 삭제하지 않습니다.",
                    "The Agent scope in the cleanup record differs. Nothing is deleted.",
                ))
        elif kind in allowed:
            if (record.get("name"), record.get("url")) != allowed[kind]:
                raise LabError(text(
                    "정리 기록의 Search/프로젝트 범위가 다릅니다. 삭제하지 않습니다.",
                    "The Search/project scope in the cleanup record differs. Nothing is deleted.",
                ))
        else:
            raise LabError(text(
                f"자동 정리를 지원하지 않는 기록 유형: {kind}",
                f"Record type not supported by automatic cleanup: {kind}",
            ))
        actions.append(dict(record))
    for metadata_path in (artifacts_dir() / "runs").glob("*/metadata.json"):
        metadata = read_json(metadata_path)
        if metadata.get("workspace_id") != state["workspace_id"]:
            continue
        if metadata.get("project_endpoint") != config.project_endpoint:
            raise LabError(text(
                "응답 정리 대상의 프로젝트 endpoint가 다릅니다.",
                "The project endpoint of the response cleanup target differs.",
            ))
        outputs_path = metadata_path.parent / "outputs.jsonl"
        if not outputs_path.exists():
            continue
        for row in read_jsonl(outputs_path):
            identifier = row.get("response_id")
            if identifier:
                if not re.fullmatch(r"resp_[A-Za-z0-9_-]+", identifier):
                    raise LabError(text(
                        "알 수 없는 response ID 형식입니다. 자동 삭제하지 않습니다.",
                        "Unrecognized response ID format. Not deleted automatically.",
                    ))
                actions.append({"kind": "response", "id": identifier})
    order = {"response": 0, "agent_version": 1, "project_connection": 2, "knowledge_base": 3, "knowledge_source": 4, "search_index": 5}
    actions.sort(key=lambda action: order[action["kind"]])
    return {
        "mode": "LOCAL_PLAN_ONLY",
        "project_id": config.project_id,
        "prefix": config.prefix,
        "workspace_id": state["workspace_id"],
        "actions": actions,
        "never_deleted": [
            "resource group", "Foundry resource/project", "shared model deployments",
            "Azure AI Search service", "Application Insights/Log Analytics", "RBAC assignments",
        ],
        "manual_follow_up": [
            "Managed evaluation jobs/files are retained for evidence; inspect recorded IDs and remove in the portal if required.",
            "SFT/FDE training jobs, uploaded training files and model deployments require their own scoped cleanup.",
            "An empty stable agent entry may remain after deleting its recorded versions; inspect ownership before removing it in the portal.",
        ],
    }

읽기용 원문이며 별도로 실행하는 명령이 아닙니다. 지정한 함수·객체 전체를 원래 메시지까지 그대로 표시합니다. 의존 함수와 import는 연결한 전체 파일에서 확인합니다. 실습 명령만 한 번 실행합니다.

전체 구현 파일 · 93–193 · GitHub에서 보기

def cleanup(config: Config, *, confirm_prefix: str | None = None) -> dict:
    plan = cleanup_plan(config)
    if confirm_prefix is None:
        return plan
    if confirm_prefix != config.prefix:
        raise LabError(text(
            "--confirm-prefix가 현재 LAB_PREFIX와 다릅니다. 삭제하지 않았습니다.",
            "--confirm-prefix differs from the current LAB_PREFIX. Nothing was deleted.",
        ))
    with credential_for(config) as credential:
        with AIProjectClient(endpoint=config.project_endpoint, credential=credential) as project:
            with project.get_openai_client(max_retries=0, timeout=60.0) as client:
                search = JsonHttp(credential, scope=SEARCH_SCOPE, allowed_origin=config.search_endpoint)
                arm = JsonHttp(credential, scope=ARM_SCOPE, allowed_origin="https://management.azure.com")
                inspected = []
                for action in plan["actions"]:
                    item = {**action, "exists": True}
                    kind = action["kind"]
                    if kind == "agent_version":
                        try:
                            remote = project.agents.get_version(agent_name=action["name"], agent_version=action["version"])
                        except ResourceNotFoundError:
                            item["exists"] = False
                        else:
                            if (remote.metadata or {}).get("workspace") != plan["workspace_id"]:
                                raise LabError(text(
                                    "원격 에이전트 소유 표식이 다릅니다. 아무 대상도 삭제하지 않았습니다.",
                                    "The remote Agent ownership marker differs. Nothing was deleted.",
                                ))
                    elif kind == "response":
                        try:
                            response = client.responses.retrieve(action["id"])
                        except NotFoundError:
                            item["exists"] = False
                        else:
                            if (response.metadata or {}).get("lab_workspace") != plan["workspace_id"]:
                                raise LabError(text(
                                    "원격 응답 소유 표식이 다릅니다. 아무 대상도 삭제하지 않았습니다.",
                                    "The remote response ownership marker differs. Nothing was deleted.",
                                ))
                    else:
                        transport = arm if kind == "project_connection" else search
                        try:
                            remote = transport.request("GET", action["url"])
                        except CloudRequestError as exc:
                            if exc.status != 404:
                                raise
                            item["exists"] = False
                        else:
                            if action.get("etag") and remote.etag and action["etag"] != remote.etag:
                                raise LabError(text(
                                    "생성 후 원격 리소스 구성이 변경되었습니다. 정리 대상을 직접 검토해야 합니다.",
                                    "The remote resource configuration changed after creation. Review the cleanup targets yourself.",
                                ))
                            item["current_etag"] = remote.etag
                    inspected.append(item)
                completed = []
                for action in inspected:
                    if not action["exists"]:
                        completed.append({**action, "result": "already_absent"})
                        continue
                    kind = action["kind"]
                    if kind == "response":
                        client.responses.delete(action["id"])
                    elif kind == "agent_version":
                        project.agents.delete_version(agent_name=action["name"], agent_version=action["version"])
                    else:
                        transport = arm if kind == "project_connection" else search
                        transport.request("DELETE", action["url"], etag=action.get("current_etag"))
                    completed.append({**action, "result": "delete_request_succeeded"})
                    save_json(artifacts_dir() / "cleanup.json", {**plan, "mode": "DELETING", "completed": completed})
                remaining = []
                for action in inspected:
                    kind = action["kind"]
                    if kind == "response":
                        try:
                            client.responses.retrieve(action["id"])
                        except NotFoundError:
                            continue
                    elif kind == "agent_version":
                        try:
                            project.agents.get_version(agent_name=action["name"], agent_version=action["version"])
                        except ResourceNotFoundError:
                            continue
                    else:
                        transport = arm if kind == "project_connection" else search
                        try:
                            transport.request("GET", action["url"])
                        except CloudRequestError as exc:
                            if exc.status == 404:
                                continue
                            raise
                    remaining.append(action)
    result = {
        **plan,
        "mode": "OWNED_OBJECTS_ABSENT" if not remaining else "DELETION_PENDING",
        "completed": completed,
        "remaining": remaining,
    }
    save_json(artifacts_dir() / "cleanup.json", result)
    return result