Skip to main content ホーム クリエイター azure contentflow contentflow-custom-executor
contentflow-custom-executor Create custom ContentFlow executors for document processing pipelines. Use when building new executors, extending BaseExecutor/ParallelExecutor/InputExecutor, working with the Content data model, registering executors in executor_catalog.yaml, or implementing process_input/process_content_item/crawl methods. Triggers on custom executor, new executor, ContentFlow executor, pipeline step, Content model, executor catalog, process_input, process_content_item, BaseExecutor, ParallelExecutor, InputExecutor.
インストールへ移動 Skills Marketplace コミュニティが作成したAIスキルを発見・探索
Codex または Claude でインストール この Prompt をコピーして Codex、Claude、または他のアシスタントに貼り付けると、Skill ページを確認してインストールできます。
直接コマンドでは確認用 Prompt が省略されます。実行前にソースを確認してください。
npx skills add https://github.com/Azure/contentflow --skill contentflow-custom-executorコマンドは1行のまま表示されます。コピー前に横へスクロールして全体を確認してください。
ローカルで確認しますか？SkillsMP が現在取得できるファイルをダウンロードできます。
Zipをダウンロード ダウンロード中... name contentflow-custom-executor description Create custom ContentFlow executors for document processing pipelines. Use when building new executors, extending BaseExecutor/ParallelExecutor/InputExecutor, working with the Content data model, registering executors in executor_catalog.yaml, or implementing process_input/process_content_item/crawl methods. Triggers on custom executor, new executor, ContentFlow executor, pipeline step, Content model, executor catalog, process_input, process_content_item, BaseExecutor, ParallelExecutor, InputExecutor.
ContentFlow Custom Executor Development
Build custom pipeline executors for the ContentFlow document processing framework.
Architecture Overview
ContentFlow executors follow a hierarchical inheritance structure built on the Agent Framework's Executor pattern:
Executor (Agent Framework)
├── BaseExecutor — Simple transformations, single/list content processing
│ ├── ParallelExecutor — Concurrent processing of multiple content items
│ │ └── AzureOpenAIAgentExecutor — AI-powered content processing
│ ├── InputExecutor — Content discovery/crawling from external sources
│ └── [Your Custom Executor]
Content Data Model
All executors work with the Content Pydantic model, defined in contentflow-lib/contentflow/models/_content.py:
from contentflow.models import Content, ContentIdentifier, ExecutorLogEntry
class ContentIdentifier (BaseModel ):
canonical_id: str
unique_id: str
source_name: str | None
str
None
str
None
str
None
str
None
dict
None
class
Content
BaseModel
id
dict
str
Any
dict
str
Any
List
Key data access patterns:
content.data["field_name"] — read/write main data fieldscontent.summary_data["key"] — read/write summary-level datacontent.id.canonical_id — get the content's canonical identifierself.try_extract_nested_field_from_content(content, "nested.field.path") — extract nested fields using dot notation
Base Classes
1. BaseExecutor When to use: Simple transformations, data validation, filtering, single-document processing.
import logging
from typing import Dict , Any , Optional , Union , List
from agent_framework import WorkflowContext, handler
from contentflow.models import Content
from contentflow.executors.base import BaseExecutor
Required method: process_input(self, input, ctx) -> Union[Content, List[Content]]
logger = logging.getLogger("contentflow.executors.my_executor" )
class MyExecutor (BaseExecutor ):
"""
Brief description of what this executor does.
Configuration (settings dict):
- my_setting (str): Description. Default: "default"
- another_setting (int): Description. Default: 10
"""
def __init__ (self, id : str , settings: Optional [Dict [str , Any ]] = None , **kwargs ):
super ().__init__(id =id , settings=settings, **kwargs)
self .my_setting = self .get_setting("my_setting" , default="default" )
self .another_setting = self .get_setting("another_setting" , default=10 )
async def process_input (
self,
input : Union [Content, List [Content]],
ctx: WorkflowContext[Union [Content, List [Content]], Union [Content, List [Content]]]
) -> Union [Content, List [Content]]:
if isinstance (input , list ):
return [await self ._process_single(item) for item in input ]
return await self ._process_single(input )
async def _process_single (self, content: Content ) -> Content:
content.data["result" ] = "processed"
return content
2. ParallelExecutor When to use: Concurrent processing of multiple items, API calls, AI model inference, I/O-bound operations.
from contentflow.executors.parallel_executor import ParallelExecutor
Required method: process_content_item(self, content: Content) -> Content
Built-in settings: max_concurrent (default: 5), timeout_secs (default: 300), continue_on_error (default: True)
class MyParallelExecutor (ParallelExecutor ):
def __init__ (self, id : str , settings: Optional [Dict [str , Any ]] = None , **kwargs ):
super ().__init__(id =id , settings=settings, **kwargs)
self .api_endpoint = self .get_setting("api_endpoint" , required=True )
async def process_content_item (self, content: Content ) -> Content:
result = await call_api(self .api_endpoint, content.data)
content.data["api_result" ] = result
return content
3. InputExecutor When to use: Content discovery/crawling from external sources, paginated retrieval, incremental updates.
from contentflow.executors.input_executor import InputExecutor
Required methods: crawl(self, checkpoint_timestamp, continuation_token) and process_input(self, input, ctx)
Built-in settings: polling_interval_seconds (default: 300), max_results (default: 1000), batch_size (default: 100)
class MyCrawlerExecutor (InputExecutor ):
def __init__ (self, id : str , settings: Optional [Dict [str , Any ]] = None , **kwargs ):
super ().__init__(id =id , settings=settings, **kwargs)
self .source_url = self .get_setting("source_url" , required=True )
async def crawl (self, checkpoint_timestamp=None , continuation_token=None ):
items = await fetch_from_source(self .source_url, since=checkpoint_timestamp)
contents = [
Content(
id =ContentIdentifier(canonical_id=item["id" ], unique_id=item["id" ]),
data=item
)
for item in items
]
return contents, None
async def process_input (self, input , ctx ):
return input
4. AzureOpenAIAgentExecutor (AI-Powered) When to use: AI/LLM-powered content processing (summarization, entity extraction, classification, etc.)
from contentflow.executors.azure_openai_agent_executor import AzureOpenAIAgentExecutor
Key pattern: Override __init__ to set specialized instructions, input_field, output_field in settings, then call super().__init__().
Built-in settings: endpoint, deployment_name, credential_type, input_field, output_field, instructions, parse_response_as_json, temperature, max_tokens
Template (follows the SummarizationExecutor/EntityExtractionExecutor pattern):
class MyAIExecutor (AzureOpenAIAgentExecutor ):
def __init__ (self, id : str , settings: Optional [Dict [str , Any ]] = None , **kwargs ):
settings = settings or {}
instructions = "You are an expert at [task]. [Detailed instructions...]"
if "input_field" not in settings:
settings["input_field" ] = "text"
if "output_field" not in settings:
settings["output_field" ] = "my_result"
settings["instructions" ] = instructions
settings["parse_response_as_json" ] = True
super ().__init__(id =id , settings=settings, **kwargs)
async def process_content_item (self, content: Content ) -> Content:
content = await super ().process_content_item(content)
return content
Configuration Management
Setting Resolution
Use self.get_setting("key", default=value) for optional settings
Use self.get_setting("key", required=True) for required settings
Environment variables: "${ENV_VAR_NAME}" syntax is auto-resolved
Access raw settings: self.settings.get("key")
Nested field extraction: self.try_extract_nested_field_from_content(content, "field.subfield")
Common Base Settings (inherited) All executors automatically support:
enabled (bool, default: True)condition (str, default: None) — condition expression evaluated per content itemfail_pipeline_on_error (bool, default: False)debug_mode (bool, default: False)
Handler Pattern The @handler decorator on handle_content in BaseExecutor is the entry point called by the Agent Framework workflow engine. You do NOT need to implement a handler — just implement the abstract methods (process_input, process_content_item, or crawl). The base handler manages:
Checking if executor is enabled
Evaluating conditions per content item
Calling your process_input() method
Error handling (fail vs. pass-through)
Sending processed content downstream via ctx.send_message() and ctx.yield_output()
Registering in executor_catalog.yaml After creating your executor, register it in contentflow-lib/executor_catalog.yaml:
- id: my_custom_executor
name: "My Custom Executor"
description: "Description of what it does"
module_path: contentflow.executors.my_custom_executor
class_name: MyCustomExecutor
tags: [custom , processing ]
category: "transform"
version: "1.0"
settings_schema:
enabled:
type: boolean
title: "Enabled"
description: "Enable or disable this executor"
required: false
default: true
ui_component: "checkbox"
condition:
type: string
title: "Condition"
description: "Condition to evaluate for each content item"
required: false
default: null
ui_component: "textarea"
fail_pipeline_on_error:
type: boolean
title: "Fail Pipeline On Error"
description: "Fail the entire pipeline on error"
required: false
default: false
ui_component: "checkbox"
debug_mode:
type: boolean
title: "Debug Mode"
description: "Enable debug logging"
required: false
default: false
ui_component: "checkbox"
my_setting:
type: string
title: "My Setting"
description: "What this setting does"
required: true
default: null
ui_component: "input"
ui_metadata:
icon: "sparkles"
description_short: "Brief one-liner"
description_long: "Detailed description for docs"
Settings Schema Types
string → ui_component: "input", "textarea", "password", "select"integer / number → ui_component: "number" (supports min, max)boolean → ui_component: "checkbox"For dropdowns: add options: ["opt1", "opt2"] with ui_component: "select"
Categories
input — Content discovery/crawling (InputExecutor subclasses)extract — Content extraction from documentstransform — Data transformation and chunkinganalyse — AI analysis and processingoutput — Writing results to destinations
Workflow YAML Usage Once registered, executors are used in workflow YAML files:
executors:
- id: my_step
type: my_custom_executor
settings:
my_setting: "value"
debug_mode: true
Error Handling Patterns Retry logic (for ParallelExecutor):
async def process_content_item (self, content: Content ) -> Content:
for attempt in range (self .max_retries):
try :
return await self ._do_work(content)
except Exception as e:
if attempt < self .max_retries - 1 :
await asyncio.sleep(self .retry_delay * (2 ** attempt))
else :
raise
async def _process_single (self, content: Content ) -> Content:
try :
result = await self ._do_work(content)
content.data["result" ] = result
except SpecificError as e:
logger.warning(f"Non-critical error for {content.id .canonical_id} : {e} " )
content.data["result" ] = None
content.data["error" ] = str (e)
return content
Testing import pytest
from contentflow.models import Content, ContentIdentifier
@pytest.mark.asyncio
async def test_my_executor ():
executor = MyExecutor(id ="test" , settings={"my_setting" : "value" })
content = Content(
id =ContentIdentifier(canonical_id="test-1" , unique_id="test-1" ),
data={"text" : "Hello world" }
)
result = await executor.process_input(content, None )
assert result.data["result" ] == "expected"
Existing Executor Reference Key executor files in contentflow-lib/contentflow/executors/:
File Base Class Purpose base.py— BaseExecutor with handler, settings, conditions parallel_executor.pyBaseExecutor ParallelExecutor with concurrency control input_executor.pyBaseExecutor InputExecutor with crawl/pagination azure_openai_agent_executor.pyParallelExecutor AI agent processing summarization_executor.pyAzureOpenAIAgentExecutor Text summarization entity_extraction_executor.pyAzureOpenAIAgentExecutor NER extraction sentiment_analysis_executor.pyAzureOpenAIAgentExecutor Sentiment analysis content_classifier_executor.pyAzureOpenAIAgentExecutor Content classification web_scraping_executor.pyBaseExecutor Web scraping with Playwright pdf_extractor.pyParallelExecutor PDF content extraction recursive_text_chunker_executor.pyBaseExecutor Text chunking azure_blob_input_discovery.pyInputExecutor Azure Blob crawling azure_blob_output_executor.pyBaseExecutor Write to Azure Blob executor_registry.py— Dynamic executor loading from catalog executor_config.py— Pydantic models for catalog config
Checklist When creating a custom executor:
