본문 바로가기
WIKI 기술 지식 베이스

Processor chains

원문 보기 위키 갱신

출처: 문서

본문

프로세서 체인

3.3에 도입됨 {: .label .label-purple }

프로세서 체인은 입력 데이터와 출력 데이터를 모두 처리할 수 있는 유연한 데이터 변환 파이프라인을 제공해요. 여러 프로세서를 체인으로 연결하면 한 프로세서의 출력이 다음 프로세서의 입력이 되는 순차적 변환을 만들 수 있어요.

프로세서를 활용하면 다음과 같은 작업을 할 수 있어요:

  • 데이터 형식 변환: 서로 다른 데이터 구조(문자열, JSON, 배열) 간에 변환해요.
  • 특정 정보 추출: JSONPath나 정규식 패턴을 사용해 필요한 데이터를 추출해요.
  • 콘텐츠 정리 및 필터링: 불필요한 필드를 제거하거나 포매팅 규칙을 적용해요.
  • 데이터 표준화: 서로 다른 컴포넌트 전반에서 일관된 데이터 형식을 보장해요.

프로세서는 배열에 나타난 순서대로 실행돼요. 각 프로세서는 이전 프로세서의 출력을 입력으로 받아요. {: .note}

프로세서 체인은 ML 워크플로를 위해 특별히 설계됐으며, ingest 파이프라인과 search 파이프라인의 프로세서와는 차이가 있어요:

  • Ingest 파이프라인: OpenSearch에 인덱싱하는 동안 문서를 변환해요.
  • Search 파이프라인: 검색 작업 중에 쿼리와 검색 결과를 변환해요.
  • 프로세서 체인: ML Commons 워크플로(agent 도구, 모델 입력/출력) 내에서 데이터를 변환해요.

프로세서 체인은 모델 응답 정리, LLM 출력에서 구조화된 데이터 추출, 모델 추론용 입력 준비 같은 AI/ML 사용 사례에 맞춘 특화된 데이터 변환 기능을 제공해요.

설정

프로세서는 다양한 컨텍스트에서 설정할 수 있어요:

  • 도구 출력: 도구의 parameters 섹션에 output_processors 배열을 추가해요.
  • 모델 출력: _predict 호출 시 모델의 parameters 섹션에 output_processors 배열을 추가해요.
  • 모델 입력: _predict 호출의 모델 parameters 섹션에 input_processors 배열을 추가해요.

전체 예제는 에이전트 사용 예제와 모델 사용 예제를 참고하세요.

지원되는 프로세서 타입

다음 표는 지원되는 모든 프로세서를 정리한 거예요.

프로세서 설명
conditional 조건에 따라 서로 다른 프로세서 체인을 적용해요.
extract_json 텍스트 문자열에서 JSON 객체나 배열을 추출해요.
for_each 배열 요소를 순회하며 각 요소에 프로세서 체인을 적용해요.
jsonpath_filter JSONPath 표현식을 사용해 데이터를 추출해요.
process_and_set 입력에 프로세서 체인을 적용하고 결과를 지정된 JSONPath 위치에 설정해요.
regex_capture 정규식 매치에서 특정 그룹을 캡처해요.
regex_replace 정규식 패턴을 사용해 텍스트를 치환해요.
remove_jsonpath JSONPath를 사용해 JSON 객체에서 필드를 제거해요.
set_field 필드에 지정된 정적 값을 설정하거나 다른 필드의 값을 복사해요.
to_string 입력을 JSON 문자열 표현으로 변환해요.

Conditional

조건에 따라 서로 다른 프로세서 체인을 적용해요.

파라미터:

  • path (string, optional): 조건 평가에 사용할 값을 추출하는 JSONPath 표현식이에요.
  • routes (array, required): 조건-프로세서 매핑의 배열이에요.
  • default (array, optional): 조건이 하나도 맞지 않을 때 사용할 기본 프로세서예요.

지원되는 조건:

  • 정확한 값 일치: "value"
  • 숫자 비교: ">10", "<5", ">=", "<=", "==5"
  • 존재 여부 확인: "exists", "null", "not_exists"
  • 정규식 매칭: "regex:pattern"
  • 텍스트 포함: "contains:substring"

설정 예제:

{
  "type": "conditional",
  "path": "$.status",
  "routes": [
    {
      "green": [
        {"type": "regex_replace", "pattern": "status", "replacement": "healthy"}
      ]
    },
    {
      "red": [
        {"type": "regex_replace", "pattern": "status", "replacement": "unhealthy"}
      ]
    }
  ],
  "default": [
    {"type": "regex_replace", "pattern": "status", "replacement": "unknown"}
  ]
}

입력 예제:

{"index": "test-index", "status": "green", "docs": 100}

출력 예제:

{"index": "test-index", "healthy": "green", "docs": 100}

extract_json

텍스트 문자열에서 JSON 객체나 배열을 추출해요.

파라미터:

  • extract_type (string, optional): 추출할 JSON 타입이에요. "object", "array", "auto" 중 하나예요. 기본값은 "auto"예요.
  • default (any, optional): JSON 추출이 실패했을 때 사용할 기본값이에요.

설정 예제:

{
  "type": "extract_json",
  "extract_type": "object",
  "default": {}
}

입력 예제:

"The result is: {\"status\": \"success\", \"count\": 5} - processing complete"

출력 예제:

{"status": "success", "count": 5}

for_each

배열 요소를 순회하며 각 요소에 프로세서 체인을 적용해요. 누락된 필드 추가, 콘텐츠 필터링, 데이터 구조 정규화처럼 배열 요소를 일관되게 변환할 때 유용해요.

파라미터:

  • path (string, required): 순회할 배열을 가리키는 JSONPath 표현식이에요. 배열 요소에는 반드시 [*] 표기법을 사용해야 해요.
  • processors (array, required): 각 배열 요소에 적용할 프로세서 설정 목록이에요.

동작 방식:

  • 각 요소는 설정된 프로세서 체인으로 독립적으로 처리돼요.
  • 프로세서 체인의 출력이 원래 요소를 대체해요.
  • 경로가 존재하지 않거나 배열을 가리키지 않으면 입력이 그대로 반환돼요.
  • 요소 처리가 실패하면 원래 요소가 유지돼요.

설정 예제:

{
  "type": "for_each",
  "path": "$.items[*]",
  "processors": [
    {
      "type": "set_field",
      "path": "$.processed",
      "value": true
    }
  ]
}

입력 예제:

{
  "items": [
    {"name": "item1", "value": 10},
    {"name": "item2", "value": 20}
  ]
}

출력 예제:

{
  "items": [
    {"name": "item1", "value": 10, "processed": true},
    {"name": "item2", "value": 20, "processed": true}
  ]
}

jsonpath_filter

JSONPath 표현식을 사용해 데이터를 추출해요.

파라미터:

  • path (string, required): 데이터를 추출하는 데 사용하는 JSONPath 표현식이에요.
  • default (any, optional): 경로를 찾지 못했을 때 사용할 기본값이에요.

설정 예제:

{
  "type": "jsonpath_filter",
  "path": "$.data.items[*].name",
  "default": []
}

입력 예제:

{"data": {"items": [{"name": "item1"}, {"name": "item2"}]}}

출력 예제:

["item1", "item2"]

process_and_set

입력에 프로세서 체인을 적용하고 결과를 지정된 JSONPath 위치에 설정해요.

파라미터:

  • path (string, required): 처리된 결과를 설정할 위치를 지정하는 JSONPath 표현식이에요.
  • processors (array, required): 순차적으로 적용할 프로세서 설정 목록이에요.

경로 동작:

  • 경로가 이미 존재하면 처리된 값으로 업데이트돼요.
  • 경로가 존재하지 않으면 프로세서 체인이 경로 생성을 시도해요(단순 중첩 필드에서 동작해요).
  • 새 필드 생성이 성공하려면 부모 경로가 반드시 존재해야 해요.

설정 예제:

{
  "type": "process_and_set",
  "path": "$.summary.clean_name",
  "processors": [
    {
      "type": "to_string"
    },
    {
      "type": "regex_replace",
      "pattern": "[^a-zA-Z0-9]",
      "replacement": "_"
    }
  ]
}

입력 예제:

{"name": "Test Index!", "status": "active"}

출력 예제:

{"name": "Test Index!", "status": "active", "summary": {"clean_name": "Test_Index_"}}

regex_capture

정규식 매치에서 특정 그룹을 캡처해요. 정규식 문법에 대한 자세한 내용은 Java 정규식 문법을 참고하세요.

파라미터:

  • pattern (string, required): 캡처 그룹이 포함된 정규식 패턴이에요.
  • groups (string or array, optional): 캡처할 그룹 번호예요. "1"처럼 단일 숫자나 "[1, 2, 4]"처럼 배열로 지정할 수 있어요. 기본값은 "1"이에요.

설정 예제:

{
  "type": "regex_capture",
  "pattern": "(\\d+),(\\w+),(\\w+),([^,]+)",
  "groups": "[1, 4]"
}

입력 예제:

"1,green,open,.plugins-ml-model-group,DCJHJc7pQ6Gid02PaSeXBQ,1,0"

출력 예제:

["1", ".plugins-ml-model-group"]

regex_replace

정규식 패턴을 사용해 텍스트를 치환해요. 정규식 문법에 대한 자세한 내용은 Java 정규식 문법을 참고하세요.

파라미터:

  • pattern (string, required): 매칭에 사용할 정규식 패턴이에요.
  • replacement (string, optional): 치환 텍스트예요. 기본값은 ""예요.
  • replace_all (Boolean, optional): 모든 매치를 치환할지 첫 번째만 치환할지 결정해요. 기본값은 true예요.

설정 예제:

{
  "type": "regex_replace",
  "pattern": "^.*?\n",
  "replacement": ""
}

입력 예제:

"row,health,status,index\n1,green,open,.plugins-ml-model\n2,red,closed,test-index"

출력 예제:

"1,green,open,.plugins-ml-model\n2,red,closed,test-index"

remove_jsonpath

JSONPath를 사용해 JSON 객체에서 필드를 제거해요.

파라미터:

  • paths (array, required): 제거할 필드를 나타내는 JSONPath 표현식 배열이에요.

설정 예제:

{
  "type": "remove_jsonpath",
  "paths": ["$.sensitive_data"]
}

입력 예제:

{"name": "user1", "sensitive_data": "secret", "public_info": "visible"}

출력 예제:

{"name": "user1", "public_info": "visible"}

set_field

필드에 지정된 정적 값을 설정하거나 다른 필드의 값을 복사해요.

파라미터:

  • path (string, required): 값을 설정할 위치를 지정하는 JSONPath 표현식이에요.
  • value (any, conditionally required): 설정할 정적 값이에요. value 또는 source_path 중 하나는 반드시 제공해야 해요.
  • source_path (string, conditionally required): 값을 복사해 올 JSONPath 표현식이에요. value 또는 source_path 중 하나는 반드시 제공해야 해요.
  • default (any, optional): source_path가 존재하지 않을 때 사용할 기본값이에요. source_path와 함께만 사용돼요.

경로 동작:

  • 경로가 이미 존재하면 새 값으로 업데이트돼요.
  • 경로가 존재하지 않으면 프로세서 체인이 경로 생성을 시도해요(단순 중첩 필드에서 동작해요).
  • 새 필드 생성이 성공하려면 부모 경로가 반드시 존재해야 해요.

설정 예제(정적 값):

{
  "type": "set_field",
  "path": "$.metadata.processed_at",
  "value": "2024-03-15T10:30:00Z"
}

설정 예제(필드 복사):

{
  "type": "set_field",
  "path": "$.userId",
  "source_path": "$.user.id",
  "default": "unknown"
}

입력 예제:

{"user": {"id": 123}, "name": "John"}

출력 예제:

{"user": {"id": 123}, "name": "John", "userId": 123, "metadata": {"processed_at": "2024-03-15T10:30:00Z"}}

to_string

입력을 JSON 문자열 표현으로 변환해요.

파라미터:

  • escape_json (Boolean, optional): JSON 문자를 이스케이프할지 여부예요. 기본값은 false예요.

설정 예제:

{
  "type": "to_string",
  "escape_json": true
}

입력 예제:

{"name": "test", "value": 123}

출력 예제:

"{\"name\":\"test\",\"value\":123}"

에이전트 사용 예제

다음 예제는 에이전트와 함께 프로세서 체인을 사용하는 방법을 보여줘요.

1단계: 출력 프로세서가 포함된 flow agent 등록

POST /_plugins/_ml/agents/_register
{
  "name": "Index Summary Agent",
  "type": "flow",
  "description": "Agent that provides clean index summaries",
  "tools": [
    {
      "type": "ListIndexTool",
      "parameters": {
        "output_processors": [
          {
            "type": "regex_replace",
            "pattern": "^.*?\n",
            "replacement": ""
          },
          {
            "type": "regex_capture",
            "pattern": "(\\d+,\\w+,\\w+,([^,]+))"
          }
        ]
      }
    }
  ]
}

{% include copy-curl.html %}

2단계: 에이전트 실행

이전 단계에서 반환된 agent_id를 사용해요:

POST /_plugins/_ml/agents/{agent_id}/_execute
{
  "parameters": {
    "question": "List the indices"
  }
}

{% include copy-curl.html %}

출력 프로세서가 없으면 ListIndexTool이 헤더와 추가 컬럼이 포함된 장황한 CSV 출력을 그대로 반환해요:

row,health,status,index,uuid,pri,rep,docs.count,docs.deleted,store.size,pri.store.size
1,green,open,.plugins-ml-model-group,DCJHJc7pQ6Gid02PaSeXBQ,1,0,1,0,12.7kb,12.7kb
2,green,open,.plugins-ml-memory-message,6qVpepfRSCi9bQF_As_t2A,1,0,7,0,53kb,53kb
3,green,open,.plugins-ml-memory-meta,LqP3QMaURNKYDZ9p8dTq3Q,1,0,2,0,44.8kb,44.8kb

출력 프로세서는 다음 과정을 거쳐 장황한 CSV 출력을 깔끔하고 읽기 쉬운 형식으로 변환해요:

  1. regex_replace: CSV 헤더 행을 제거해요.
  2. regex_capture: 필수 정보(행 번호, health, status, 인덱스 이름)만 추출해요.

출력 프로세서를 사용하면 에이전트가 필수 인덱스 정보만 담긴 깔끔하게 포맷된 데이터를 반환해요:

1,green,open,.plugins-ml-model-group
2,green,open,.plugins-ml-memory-message
3,green,open,.plugins-ml-memory-meta

모델 사용 예제

다음 예제는 Predict API 호출 중에 모델과 함께 프로세서 체인을 사용하는 방법을 보여줘요.

예제: 입력 프로세서

이 예제는 input_processors를 사용해 처리 전에 텍스트를 치환하는 방식으로 모델 입력을 수정하는 방법을 보여줘요:

POST _plugins/_ml/models/{model_id}/_predict
{
  "parameters": {
    "system_prompt": "You are a helpful assistant.",
    "prompt": "Can you summarize Prince Hamlet of William Shakespeare in around 100 words?",
    "input_processors": [
      {
        "type": "regex_replace",
        "pattern": "100",
        "replacement": "20"
      }
    ]
  }
}

{% include copy-curl.html %}

이 예제에서 regex_replace 프로세서는 모델로 전송되기 전에 프롬프트를 수정해 "100 words"를 "20 words"로 바꿔요.

예제: 출력 프로세서

이 예제는 output_processors를 사용해 모델 출력을 처리해서 JSON 데이터를 추출하고 포맷하는 방법을 보여줘요. 이 예제에서 출력 프로세서는 먼저 JSONPath를 사용해 모델 응답에서 콘텐츠를 추출해요. 그다음 텍스트 응답에서 JSON 객체를 파싱해서 추출해요:

POST _plugins/_ml/models/{model_id}/_predict
{
  "parameters": {
    "messages": [
      {
        "role": "system",
        "content": [
          {
            "type": "text",
            "text": "${parameters.system_prompt}"
          }
        ]
      },
      {
        "role": "user",
        "content": [
          {
            "type": "text",
            "text": "Can you convert this into a json object: user name is Bob, he likes swimming"
          }
        ]
      }
    ],
    "system_prompt": "You are a helpful assistant",
    "output_processors": [
      {
        "type": "jsonpath_filter",
        "path": "$.choices[0].message.content"
      },
      {
        "type": "extract_json",
        "extract_type": "auto"
      }
    ]
  }
}

{% include copy-curl.html %}

출력 프로세서가 없으면 응답에 방대한 메타데이터와 중첩 구조가 포함된 전체 모델 출력이 그대로 담겨요:

{
  "inference_results": [
    {
      "output": [
        {
          "name": "response",
          "dataAsMap": {
            "id": "test-id",
            "object": "chat.completion",
            "created": 1.759580469E9,
            "model": "gpt-4o-mini-2024-07-18",
            "choices": [
              {
                "index": 0.0,
                "message": {
                  "role": "assistant",
                  "content": "Sure! Here is the information you provided converted into a JSON object:\n\n```json\n{\n  \"user\": {\n    \"name\": \"Bob\",\n    \"likes\": \"swimming\"\n  }\n}\n```",
                  "refusal": null,
                  "annotations": []
                },
                "logprobs": null,
                "finish_reason": "stop"
              }
            ],
            "usage": {
              "prompt_tokens": 33.0,
              "completion_tokens": 42.0,
              "total_tokens": 75.0,
              "prompt_tokens_details": {
                "cached_tokens": 0.0,
                "audio_tokens": 0.0
              },
              "completion_tokens_details": {
                "reasoning_tokens": 0.0,
                "audio_tokens": 0.0,
                "accepted_prediction_tokens": 0.0,
                "rejected_prediction_tokens": 0.0
              }
            },
            "service_tier": "default",
            "system_fingerprint": "test-fingerprint"
          }
        }
      ],
      "status_code": 200
    }
  ]
}

출력 프로세서를 사용하면 응답이 단순해져서 추출하고 파싱한 JSON 데이터만 포함해요:

{
  "inference_results": [
    {
      "output": [
        {
          "name": "response",
          "dataAsMap": {
            "user": {
              "name": "Bob",
              "likes": "swimming"
            }
          }
        }
      ],
      "status_code": 200
    }
  ]
}

더 알아보기 (Learn more)

함께 읽는 실무 이야기

우리 조직에도 적용하고 싶다면, AI 시스템 구축 서비스를 살펴보세요.