코드 실행 및 파일 교환

  • 14분

Tip

자세한 내용은 텍스트 및 이미지 탭을 참조하세요.

AI 백 엔드는 세션 풀 관리 API를 통해 동적 세션과 상호 작용합니다. 백 엔드는 각 사용자 또는 대화에 애플리케이션 제어 식별자를 할당하기 전에 Microsoft Entra ID로 인증합니다. 그런 다음, 권한 있는 코드와 파일만 격리된 환경으로 보냅니다.

Note

이 단원의 Python 예제에서는 requests, azure-identity, langchain-azure-dynamic-sessions를 사용하여 통합 패턴을 보여 줍니다. 프로덕션 환경에서 사용하기 전에 애플리케이션의 HTTP 클라이언트, 오케스트레이션 프레임워크, 유효성 검사 정책 및 현재 코드 인터프리터 데이터 평면 API에 조각을 조정하세요.

백엔드 인증

모든 관리 API 요청에는 Microsoft Entra Bearer 토큰이 필요합니다. 호출 ID에는 세션 풀에서 Azure ContainerApps 세션 실행기 역할이 필요하며 토큰의 대상은 https://dynamicsessions.io여야 합니다. 관리형 ID는 백 엔드가 Azure에서 실행되면 암호 없는 인증을 제공합니다.

ID 엔터티가 관련 없는 풀에서 코드를 실행할 수 없도록 세션 풀 범위에서 역할을 할당할 수 있습니다. 다음 명령은 백엔드 ID의 보안 주체 ID 및 세션 풀 리소스 ID를 이미 알고 있다고 가정합니다.

az role assignment create \
    --role "Azure ContainerApps Session Executor" \
    --assignee <BACKEND_PRINCIPAL_ID> \
    --scope <SESSION_POOL_RESOURCE_ID>

애플리케이션 코드는 로컬 개발 및 Azure 호스팅 환경 모두에서 DefaultAzureCredential를 사용할 수 있습니다. 자격 증명은 소스 코드에 애플리케이션 비밀을 저장하지 않고 로컬에서 사용 가능한 개발자 자격 증명 및 Azure의 관리형 ID를 선택합니다.

from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
access_token = credential.get_token(
    "https://dynamicsessions.io/.default"
).token

애플리케이션이 토큰을 무기한 캐시하지 말고 호출을 준비할 때 토큰을 요청하세요. DefaultAzureCredential 는 토큰 캐싱 및 갱신을 처리하고 코드는 다음 요청에 유효한 토큰을 가져옵니다.

보안 세션 식별자 만들기

세션 식별자는 요청을 기존 세션으로 라우팅하거나 풀이 새 세션을 할당하게 합니다. 식별자를 다시 사용하면 관련 호출에서 임시 파일 및 상태가 유지됩니다. 식별자는 권한 있는 토큰과 결합될 때 해당 상태에 대한 액세스 권한을 부여하므로 백엔드는 이를 민감한 정보로 처리해야 합니다.

인증된 각 사용자 또는 에이전트 대화에 대해 서버에서 예측할 수 없는 식별자를 생성합니다. 연관 관계를 서버 쪽 애플리케이션 상태로 유지하고 클라이언트의 임의 식별자를 허용하지 않습니다. 임의 UUID는 하이픈 처리된 표현이 동적 세션 식별자 형식을 충족하므로 적합합니다.

from uuid import uuid4


def create_session_identifier() -> str:
    return str(uuid4())

식별자만으로는 호출자를 인증하지 않습니다. 백 엔드는 저장된 연결을 사용하기 전에 사용자에게 권한을 부여하고 자체 토큰으로 모든 풀 요청을 인증합니다. 이 디자인은 사용자가 요청 매개 변수를 변경하여 다른 대화의 환경을 대상으로 하는 것을 방지합니다.

실행을 위한 코드 제출

코드 인터프리터 executions 엔드포인트는 인라인 코드를 수락하고 실행 응답을 반환합니다. 프로덕션 AI 애플리케이션에서 오케스트레이션 계층은 코드를 제공할 수 있지만 백엔드는 제출 전에 요청이 해당 정책을 준수하는지 검증해야 합니다. 격리는 신뢰할 수 없는 코드의 영향을 줄이지만 요청된 작업이 적절한지 여부는 결정하지 않습니다.

다음 헬퍼는 인증 헤더 및 공통 쿼리 매개변수를 만듭니다. DefaultAzureCredential에 각 작업마다 유효한 토큰을 요청하고 세션 식별자는 애플리케이션이 제어합니다.

from azure.identity import DefaultAzureCredential

API_VERSION = "2025-10-02-preview"
TOKEN_SCOPE = "https://dynamicsessions.io/.default"


class DynamicSessionClient:
    def __init__(
        self,
        pool_management_endpoint: str,
        session_identifier: str,
    ) -> None:
        self.endpoint = pool_management_endpoint.rstrip("/")
        self.identifier = session_identifier
        self.credential = DefaultAzureCredential()

    def _headers(self) -> dict[str, str]:
        token = self.credential.get_token(TOKEN_SCOPE).token
        return {"Authorization": f"Bearer {token}"}

    def _params(self) -> dict[str, str]:
        return {
            "api-version": API_VERSION,
            "identifier": self.identifier,
        }

다음 메서드는 동기 Python 코드를 전송합니다. raise_for_status() 는 인증, 유효성 검사, 용량 및 서비스 오류가 성공적인 실행처럼 보이지 않도록 합니다. 호출자는 Python 코드에서 발생한 오류에 대해 반환된 실행 데이터를 검사해야 합니다.

import requests


def execute_python(
    client: DynamicSessionClient,
    code: str,
) -> dict:
    response = requests.post(
        f"{client.endpoint}/executions",
        headers={
            **client._headers(),
            "Content-Type": "application/json",
        },
        params=client._params(),
        json={
            "properties": {
                "codeInputType": "inline",
                "executionType": "synchronous",
                "code": code,
            }
        },
        timeout=30,
    )
    response.raise_for_status()
    return response.json()

기능이 진화하는 동안 데이터 평면 API 버전이 변경 될 수 있습니다. 요청 메서드에 분산하는 대신 구성에서 버전을 유지한 다음 업데이트할 때 응답 계약 사양을 테스트합니다.

세션을 사용하여 파일 교환

파일 작업을 사용하면 코드가 사용자 업로드를 처리하고 차트 또는 변환된 문서와 같은 아티팩트를 생성할 수 있습니다. 백 엔드는 현재 작업에 대해 권한이 부여된 파일만 업로드합니다. 코드 인터프리터 세션은 업로드된 파일을 /mnt/data에 배치하며, 그곳에 있는 파일은 동일한 식별자를 가진 후속 실행에서 액세스할 수 있습니다.

다음 메서드는 다중 파트 요청을 사용하여 파일을 업로드합니다. 컨텍스트 관리자는 요청이 실패하더라도 로컬 파일을 닫고 명시적 시간 제한으로 인해 백 엔드가 무기한 대기하지 못하게 됩니다.

from pathlib import Path

import requests


def upload_file(
    client: DynamicSessionClient,
    file_path: Path,
) -> dict:
    with file_path.open("rb") as source:
        response = requests.post(
            f"{client.endpoint}/files",
            headers=client._headers(),
            params=client._params(),
            files={
                "file": (
                    file_path.name,
                    source,
                    "application/octet-stream",
                )
            },
            timeout=30,
        )

    response.raise_for_status()
    return response.json()

코드가 출력 파일을 만든 후 백 엔드는 세션 파일을 나열하고 선택한 아티팩트를 다운로드할 수 있습니다. 지원되는 파일 이름에 예약된 URL 문자가 포함될 수 있으므로 서비스 쪽 파일 이름을 하나의 URL 경로 세그먼트로 인코딩하세요. 경로 순회가 의도하지 않은 파일을 덮어쓸 수 있으므로 유효성을 검사하지 않고 신뢰할 수 없는 파일 이름에서 로컬 대상을 생성하지 마세요.

from urllib.parse import quote


def list_files(client: DynamicSessionClient) -> dict:
    response = requests.get(
        f"{client.endpoint}/files",
        headers=client._headers(),
        params=client._params(),
        timeout=15,
    )
    response.raise_for_status()
    return response.json()


def download_file(
    client: DynamicSessionClient,
    file_name: str,
    destination: Path,
) -> None:
    encoded_file_name = quote(file_name, safe="")
    response = requests.get(
        f"{client.endpoint}/files/{encoded_file_name}/content",
        headers=client._headers(),
        params=client._params(),
        timeout=30,
    )
    response.raise_for_status()
    destination.write_bytes(response.content)

코드 인터프리터 파일 업로드 제한은 128MB입니다. 더 큰 원본 데이터의 경우 세션이 더 작은 승인된 입력을 받도록 작업을 다시 디자인하거나 워크로드의 스토리지 요구 사항을 충족하는 사용자 지정 실행 아키텍처를 사용합니다.

작업 간에 세션을 다시 사용

관련 호출은 세션이 만료되기 전에 동일한 식별자를 포함할 때 동일한 환경을 다시 사용합니다. 이 동작을 사용하면 애플리케이션에서 문서를 한 번 업로드하고, 분석을 실행하고, 원본을 다시 업로드하지 않고도 다른 시각화를 만들 수 있습니다. 다른 식별자는 별도의 환경을 만들거나 가리킵니다.

다음 시퀀스는 하나의 클라이언트 인스턴스를 사용하므로 식별자도 하나입니다. 제공된 코드 페이로드는 AI 오케스트레이션 계층의 유효성이 검사된 출력을 나타냅니다.

from pathlib import Path

session_id = create_session_identifier()
client = DynamicSessionClient(pool_endpoint, session_id)

upload_file(client, Path("operational-data.csv"))
execution = execute_python(client, supplied_python_payload)
available_files = list_files(client)
download_file(client, "trend.png", Path("trend.png"))

세션 재사용은 영구적 저장 보장이 아니라 일시적인 편의성입니다. 승인된 결과 및 원본 데이터를 애플리케이션 제어 스토리지에 저장하여 쿨다운이 세션을 제거할 때 백 엔드가 복구할 수 있도록 합니다.

LangChain 에이전트 통합

직접 REST 액세스는 백엔드가 세션 식별자, 파일 교환 또는 API 응답을 명시적으로 제어해야 하는 경우에 적합합니다. AI 애플리케이션이 이미 LangChain을 사용하는 경우, langchain-azure-dynamic-sessions 해당 패키지는 SessionsPythonREPLTool을 동일한 코드 인터프리터 풀과 더 높은 수준의 통합으로 제공합니다. 이 도구는 애플리케이션이 각 실행 요청을 생성할 필요 없이 LangChain 에이전트의 도구 호출 워크플로에 참여할 수 있습니다.

다음 조각은 풀 관리 엔드포인트에서 도구를 만들고 에이전트에서 사용할 수 있도록 합니다. 이 도구는 애플리케이션이 인스턴스를 만들 때 임의의 세션 식별자를 생성합니다. 해당 인스턴스를 통한 호출은 세션이 만료될 때까지 동일한 세션을 다시 사용합니다.

# Code fragment - focus on creating the LangChain tool
from langchain_azure_dynamic_sessions import SessionsPythonREPLTool

python_repl = SessionsPythonREPLTool(
    pool_management_endpoint=pool_management_endpoint
)
agent_tools = [python_repl]

모든 사용자에 대해 하나의 장기 도구 인스턴스를 공유 싱글톤으로 등록하지 마세요. 관련 없는 요청이 임시 파일 또는 Python 상태를 공유하지 않도록 인증된 각 사용자 또는 대화에 대해 별도의 도구 및 에이전트 컨텍스트를 만듭니다. 애플리케이션은 여전히 사용자 권한 부여를 제어하고, 에이전트 요청의 유효성을 검사하고, Azure ContainerApps 세션 실행기 역할을 해당 ID에 할당하고, 지속성 아티팩트를 세션 외부에 저장합니다.

LangChain 어댑터는 에이전트가 Python 실행을 도구로 주로 필요로 하는 경우에 유용합니다. 직접 API 호출은 백 엔드가 실행 전에 원본 파일을 업로드하거나, 특정 아티팩트를 다운로드하거나, 식별자 매핑을 관리하거나, 전체 서비스 응답을 해석해야 하는 경우 보다 명시적인 제어를 제공합니다. 전체 에이전트 예제는 Azure Container Apps를 사용하여 LangChain에서 코드 인터프리터 세션 사용을(를) 참조하세요.

추가 리소스

이러한 리소스는 현재 인증, 실행 및 파일 API 세부 정보를 제공합니다. 백 엔드 통합을 구현할 때 이를 사용하여 요청 형식 및 보안 요구 사항을 확인할 수 있습니다.