aw

aw - Agentic Workflows for Data Preparation

An AI agent package for data preparation with a focus on loading and preparing data for various purposes (e.g., visualization with cosmograph).

Key Features: - AgenticStep protocol for building modular agents - ReAct pattern (Reason-Act-Observe) with retry logic - Three validation flavors: schema, info-dict, and functional - Code execution with safe defaults and extensibility - Loading and Preparation agents for data workflows - Cosmograph-specific validators and utilities - Orchestration for chaining multiple steps

Example

>>> from aw import load_for_cosmo
>>> df, metadata = load_for_cosmo('data.csv')
>>> from cosmograph import cosmo
>>> params = metadata['preparing']['metadata']['validation_result']['params']
>>> cosmo(df, **params)

Architecture:

• aw.base: Core protocols and configurations

• aw.validation: Three validation flavors

• aw.tools: Code execution and agent tools

• aw.utils: Helper functions and facades

• aw.loading: LoadingAgent for data ingestion

• aw.preparing: PreparationAgent for data transformation

• aw.cosmo: Cosmograph-specific validators

• aw.orchestration: Workflow management

class aw.AgentSpec(name: str, description: str = '', instructions: str = '', tools: list = <factory>, validators: list = <factory>, model: str = '', max_retries: int = 3, human_in_loop: bool = False, source_class: str = '', extra: dict = <factory>)

Normalized, format-agnostic description of an aw agent.

This is the intermediate representation that all translators consume. It captures the essential components of an agent without being tied to any specific framework.

Example

>>> from aw import LoadingAgent
>>> spec = extract_agent_spec(LoadingAgent())
>>> spec.name
'LoadingAgent'

class aw.AgenticStep(*args, **kwargs)

Protocol defining the interface for any agentic step in a workflow.

Each step follows the ReAct pattern: 1. Reason (Thought): Analyze input and context 2. Act (Action): Generate and execute code or use tools 3. Observe: Capture results and validate 4. Repeat or finish based on validation

Example

>>> class MyAgent:
...     def execute(self, input_data, context):
...         # Agent implementation
...         return result, metadata

execute(input_data: Any, context: MutableMapping[str, Any]) → tuple[ArtifactType, dict[str, Any]]

Execute the agentic step.

Parameters:

• input_data – The input to process

• context – Mutable mapping containing shared state and history

Returns:

Tuple of (artifact, metadata) where artifact is the main output and metadata contains auxiliary information

class aw.AgenticWorkflow(context: Context = None)

Orchestrates a chain of agentic steps.

Manages the execution of multiple steps in sequence, handling context/artifact passing between steps.

Example

>>> workflow = AgenticWorkflow()
>>> workflow.add_step('loading', loading_agent)
>>> workflow.add_step('preparing', preparing_agent)
>>> result = workflow.run(source_uri)

add_step(name: str, step: AgenticStep) → AgenticWorkflow

Add a step to the workflow.

Parameters:

• name – Name/identifier for the step

• step – Agent or step implementation

Returns:

Self for chaining

run(initial_input: Any) → tuple[Any, dict[str, Any]]

Execute the workflow.

Parameters:

initial_input – Input to the first step

Returns:

Tuple of (final_artifact, workflow_metadata)

run_partial(initial_input: Any, stop_after: str = None) → tuple[Any, dict[str, Any]]

Run workflow up to a specific step.

Parameters:

• initial_input – Input to first step

• stop_after – Name of step to stop after (runs all if None)

Returns:

Tuple of (artifact, metadata)

class aw.CodeInterpreterTool(allowed_modules: list[str] = None, global_context: dict = None, executor: Callable[[str, dict], ExecutionResult] = None)

Tool for executing Python code in a controlled environment.

Provides a safe default implementation using exec() with limited namespace, and allows injection of more robust backends.

Example

>>> tool = CodeInterpreterTool()
>>> result = tool("x = 5; y = x * 2; print(y)")
>>> result.success
True
>>> result.output.strip()
'10'

class aw.Context(initial_data: dict = None)

Context for sharing state and artifacts between agentic steps.

Implements MutableMapping interface for dict-like behavior while providing additional functionality for managing agent state.

Example

>>> ctx = Context()
>>> ctx['loading'] = {'df': df, 'info': {...}}
>>> ctx['preparing'] = {'df': prepared_df}

property history: list[tuple[str, Any]]

Access the history of all context updates.

snapshot() → dict[str, Any]

Create a snapshot of current context state.

Returns:

A dict copy of current context data

class aw.ExecutionResult(success: bool, output: str = '', error: str = '', traceback_str: str = '', result: Any = None, locals_dict: dict = None)

Result of code execution.

success

Whether execution succeeded without exceptions

output

Standard output captured during execution

error

Error message if execution failed

traceback

Full traceback if execution failed

result

The return value or final expression value

locals

Local variables after execution

class aw.FileSamplerTool(sample_size: int = 1024)

Tool to sample and analyze file metadata and content.

Helps agents understand what kind of data they’re dealing with before attempting to load it.

Example

>>> sampler = FileSamplerTool()
>>> info = sampler('/path/to/data.csv')
>>> info['extension']
'.csv'

class aw.GlobalConfig(llm: str | Callable[[str], str] = 'gpt-4', max_retries: int = 3, human_in_loop: bool = False)

Global configuration with cascading defaults.

Provides defaults that can be overridden at step or agent level.

Example

>>> global_cfg = GlobalConfig(llm="gpt-4", max_retries=5)
>>> step_cfg = global_cfg.override(llm="gpt-3.5-turbo")

override(**kwargs) → StepConfig

Create a StepConfig with overridden values.

Parameters:

**kwargs – Values to override from global defaults

Returns:

A new StepConfig with merged configuration

class aw.InteractiveWorkflow(context: Context = None)

Workflow with human-in-the-loop capabilities.

Pauses execution to request human input/approval at configured points.

Example

>>> workflow = InteractiveWorkflow()
>>> workflow.add_step('loading', agent, require_approval=True)
>>> result = workflow.run_interactive(source_uri)

add_step(name: str, step: AgenticStep, require_approval: bool = False) → InteractiveWorkflow

Add step with optional approval requirement.

run_interactive(initial_input: Any) → tuple[Any, dict]

Run workflow with human-in-loop checkpoints.

set_approval_callback(callback: Callable[[str, Any, dict], bool]) → None

Set callback for human approval.

Parameters:

callback – Function(step_name, artifact, metadata) -> bool Returns True to continue, False to abort

class aw.LoadingAgent(config: StepConfig = None)

Agent that loads data from various sources into pandas DataFrames.

Uses ReAct loop: 1. Thought: Analyze source (extension, sample) to choose loader 2. Action: Generate code to load data 3. Observe: Execute code and capture result/error 4. Validate: Check if result is valid DataFrame 5. Repeat or finish

Example

>>> agent = LoadingAgent()
>>> context = Context()
>>> df, metadata = agent.execute('/path/to/data.csv', context)

execute(source_uri: str, context: MutableMapping[str, Any]) → tuple[Any, dict[str, Any]]

Execute loading agent to load data from source.

Parameters:

• source_uri – URI of data source (file path or URL)

• context – Context for storing intermediate results

Returns:

Tuple of (dataframe, metadata)

class aw.PreparationAgent(config: StepConfig = None, target: str = 'generic', target_validator: Callable = None)

Agent that prepares data to meet target requirements.

Uses ReAct loop with functional validation: 1. Thought: Analyze current data state vs. requirements 2. Action: Generate transformation code 3. Observe: Execute code and capture result 4. Validate: Try to use data for its purpose (e.g., visualization) 5. Repeat or finish

Example

>>> agent = PreparationAgent(target='cosmo-ready')
>>> context = Context({'loading': {'df': df}})
>>> prepared_df, metadata = agent.execute(df, context)

execute(input_df: Any, context: MutableMapping[str, Any]) → tuple[Any, dict[str, Any]]

Execute preparation agent to transform data.

Parameters:

• input_df – Input DataFrame to prepare

• context – Context for storing intermediate results

Returns:

Tuple of (prepared_dataframe, metadata)

class aw.SafeCodeInterpreter(**kwargs)

Extra-safe code interpreter with restricted operations.

Disallows file I/O, network access, and other potentially dangerous operations.

class aw.StepConfig(llm: str | ~typing.Callable[[str], str] = 'gpt-4', validator: ~typing.Callable | ~typing.Any = None, tools: list[~typing.Callable] = <factory>, max_retries: int = 3, human_in_loop: bool = False)

Configuration for an AgenticStep.

Supports both simple objects (strings, ints) and callables for maximum flexibility with dependency injection.

llm

Either a model name string or a text-to-text chat function

Type:

str | Callable[[str], str]

validator

Validation callable or schema object

Type:

Callable | Any

tools

List of callable tools available to the agent

Type:

list[Callable]

max_retries

Maximum number of retry attempts

Type:

int

human_in_loop

Whether to require human approval/intervention

Type:

bool

Example

>>> config = StepConfig(
...     llm="gpt-4",
...     validator=lambda x: (True, {}),
...     max_retries=3
... )

resolve_llm() → Callable[[str], str]

Resolve llm to a callable function.

Returns:

A callable that takes a prompt string and returns a response string

resolve_validator() → Validator

Resolve validator to a callable.

Returns:

A callable that validates artifacts

class aw.Tool(*args, **kwargs)

Protocol for tools that agents can use.

Tools are callable objects that perform specific actions like executing code, sampling files, or calling external APIs.

Example

>>> class FileSampler:
...     def __call__(self, uri):
...         # Sample file logic
...         return {'extension': '.csv', 'sample': '...'}

class aw.ToolSpec(name: str, description: str = '', parameters: dict = <factory>, source: str = '')

Normalized description of a tool.

class aw.Validator(*args, **kwargs)

Protocol for validation functions.

Validators check if an artifact meets requirements and return both a success indicator and detailed information.

Example

>>> def is_non_empty_dataframe(df):
...     success = df is not None and len(df) > 0
...     info = {'shape': df.shape if success else None}
...     return success, info

class aw.ValidatorSpec(name: str, description: str = '', checks: list = <factory>)

Normalized description of a validator.

aw.all_validators(*validators: Callable) → Callable[[Any], tuple[bool, dict]]

Combine multiple validators - all must pass.

Parameters:

*validators – Validator functions to combine

Returns:

A validator that passes only if all validators pass

Example

>>> validate = all_validators(
...     is_dataframe,
...     has_required_columns(['x', 'y']),
...     has_no_nulls
... )

aw.any_validator(*validators: Callable) → Callable[[Any], tuple[bool, dict]]

Combine multiple validators - at least one must pass.

Parameters:

*validators – Validator functions to combine

Returns:

A validator that passes if any validator passes

aw.basic_cosmo_validator() → Callable

Basic validator that checks structural requirements only.

Does not require cosmograph to be installed.

Example

>>> validator = basic_cosmo_validator()
>>> success, info = validator(df)

aw.claude_skill_from_spec(spec: AgentSpec, *, name: str = None, description: str = None, extra_tools: list = None, disable_model_invocation: bool = False) → str

Render a Claude Code SKILL.md string from an AgentSpec.

The rendering core of to_claude_skill() — driveable by any AgentSpec (not only a live aw agent), so other tools (e.g. coact) can reuse it.

Example

>>> from aw import LoadingAgent
>>> from aw.translators import extract_agent_spec, claude_skill_from_spec
>>> spec = extract_agent_spec(LoadingAgent(), name='data-loading')
>>> '---' in claude_skill_from_spec(spec)
True

aw.compute_dataframe_info(df) → dict

Compute comprehensive info about a DataFrame.

Parameters:

df – pandas DataFrame

Returns:

Dict with shape, dtypes, null counts, sample rows, etc.

Example

>>> import pandas as pd
>>> df = pd.DataFrame({'a': [1, 2], 'b': [3, 4]})
>>> info = compute_dataframe_info(df)
>>> info['shape']
(2, 2)

aw.create_cosmo_prep_workflow(cosmo_validator: Callable = None, max_retries: int = 3) → AgenticWorkflow

Create a workflow specifically for cosmograph preparation.

Parameters:

• cosmo_validator – Custom cosmo validator (uses default if None)

• max_retries – Maximum retries for each step

Returns:

Configured workflow

Example

>>> workflow = create_cosmo_prep_workflow()
>>> prepared_df, metadata = workflow.run('data.csv')

aw.create_cosmo_validator(cosmo_function: Callable = None, required_columns: int = 2, allow_generated_params: bool = True) → Callable[[Any], tuple[bool, dict]]

Create a validator for cosmograph requirements.

This validator attempts to actually call cosmograph.cosmo() with the DataFrame to see if it’s ready for visualization.

Parameters:

• cosmo_function – The cosmograph.cosmo function (imported if None)

• required_columns – Minimum number of numeric columns needed

• allow_generated_params – Whether to auto-infer x/y columns

Returns:

Validator function following (artifact) -> (success, info) protocol

Example

>>> validator = create_cosmo_validator()
>>> success, info = validator(df)
>>> if success:
...     print(f"Can visualize with: {info['params']}")

aw.create_data_prep_workflow(loading_config: StepConfig = None, preparing_config: StepConfig = None, target: str = 'generic') → AgenticWorkflow

Factory to create a data preparation workflow.

Creates a workflow with LoadingAgent -> PreparationAgent.

Parameters:

• loading_config – Configuration for loading agent

• preparing_config – Configuration for preparing agent

• target – Target format for preparation

Returns:

Configured workflow

Example

>>> workflow = create_data_prep_workflow(target='cosmo-ready')
>>> df, metadata = workflow.run('/path/to/data.csv')

aw.create_langchain_executor() → Callable[[str, dict], ExecutionResult]

Create an executor using LangChain’s CodeInterpreter (if available).

Returns:

Executor function compatible with CodeInterpreterTool

Example

>>> try:
...     executor = create_langchain_executor()
...     tool = CodeInterpreterTool(executor=executor)
... except ImportError:
...     # LangChain not available - use default
...     tool = CodeInterpreterTool()

aw.create_loading_agent(llm: str = None, validator: Any = None, max_retries: int = 3) → LoadingAgent

Factory function to create a loading agent.

Parameters:

• llm – LLM model name or callable

• validator – Custom validator (uses default if None)

• max_retries – Maximum retry attempts

Returns:

Configured LoadingAgent

Example

>>> agent = create_loading_agent(llm="gpt-4", max_retries=5)

aw.create_oa_chat() → Callable[[str], str]

Create a chat function using the oa package.

Returns:

A callable that takes a prompt and returns a response

Example

>>> chat = create_oa_chat()
>>> response = chat("Hello!")

aw.create_openai_chat(model: str = 'gpt-4', **default_kwargs) → Callable[[str], str]

Create a text-to-text chat function using OpenAI API.

Parameters:

• model – OpenAI model name

• **default_kwargs – Default parameters for API calls

Returns:

A callable that takes a prompt and returns a response

Example

>>> chat = create_openai_chat("gpt-4")
>>> response = chat("What is 2+2?")

aw.create_preparation_agent(target: str = 'generic', validator: Callable = None, llm: str = None, max_retries: int = 3) → PreparationAgent

Factory function to create a preparation agent.

Parameters:

• target – Target format/purpose

• validator – Custom validator function

• llm – LLM model name or callable

• max_retries – Maximum retry attempts

Returns:

Configured PreparationAgent

Example

>>> agent = create_preparation_agent(
...     target='cosmo-ready',
...     validator=cosmo_validator,
...     max_retries=5
... )

aw.crewai_yaml_from_spec(spec: AgentSpec, *, name: str = None, role: str = None, goal: str = None, backstory: str = None) → dict

Render a CrewAI agent config dict from an AgentSpec.

The rendering core of to_crewai_yaml(), driveable by any AgentSpec.

aw.default_llm_factory(model: str | Callable = 'gpt-4') → Callable[[str], str]

Factory to create LLM function from various specifications.

Parameters:

model – Either a model name string, or a callable

Returns:

A callable text-to-text function

Example

>>> llm = default_llm_factory("gpt-4")
>>> llm = default_llm_factory(lambda p: f"Echo: {p}")

aw.djoin(*p)

Join two or more pathname components, inserting ‘/’ as needed. If any component is an absolute path, all previous path components will be discarded. An empty last part will result in a path that ends with a separator.

aw.extract_agent_spec(agent: Any, name: str = None) → AgentSpec

Extract a normalized AgentSpec from any aw agent.

Introspects the agent’s class, config, tools, and validators to build a format-agnostic intermediate representation.

Parameters:

• agent – An aw agent instance (LoadingAgent, PreparationAgent, etc.)

• name – Override name (defaults to class name)

Returns:

AgentSpec with all extractable information

Example

>>> from aw import LoadingAgent
>>> spec = extract_agent_spec(LoadingAgent())
>>> spec.name
'LoadingAgent'
>>> len(spec.tools) >= 1
True

aw.functional_validator(try_function: Callable[[Any], Any], success_check: Callable[[Any], bool] = None) → Callable[[Any], tuple[bool, dict]]

Create a validator that tries to use the artifact for its purpose.

This validator actually attempts to use the artifact in the way it’s intended to be used (e.g., try to visualize the data, try to train a model).

Parameters:

• try_function – Function that tries to use the artifact

• success_check – Optional function to check if result is acceptable

Returns:

A validator function

Example

>>> def try_cosmo(df):
...     return cosmograph.cosmo(df, points_x_by='x', points_y_by='y')
>>> validate = functional_validator(try_cosmo)
>>> validate(df)

aw.get_numeric_columns(df, exclude_nulls: bool = True)

Get numeric columns from DataFrame.

Parameters:

• df – pandas DataFrame

• exclude_nulls – Whether to exclude columns with null values

Returns:

Generator of column names

Example

>>> for col in get_numeric_columns(df):
...     print(col)

aw.has_attributes(**required_attrs: Any) → Callable[[Any], tuple[bool, dict]]

Validator that checks artifact has required attributes.

Example

>>> validate = has_attributes(shape=lambda s: len(s) == 2, columns=lambda c: len(c) > 0)

aw.infer_cosmo_params(df) → dict

Infer suitable cosmograph parameters from DataFrame.

Parameters:

df – pandas DataFrame

Returns:

Dict of suggested parameters for cosmograph.cosmo()

Example

>>> params = infer_cosmo_params(df)
>>> from cosmograph import cosmo
>>> cosmo(df, **params)

aw.infer_loader_from_extension(extension: str) → str

Infer pandas loader function from file extension.

Parameters:

extension – File extension (e.g., ‘.csv’, ‘.json’)

Returns:

Name of pandas loader function

Example

>>> infer_loader_from_extension('.csv')
'read_csv'
>>> infer_loader_from_extension('.xlsx')
'read_excel'

aw.infer_loader_params(extension: str, sample_text: str = None) → dict

Infer parameters for pandas loader based on file characteristics.

Parameters:

• extension – File extension

• sample_text – Optional sample of file content

Returns:

Dict of parameters to pass to loader

Example

>>> params = infer_loader_params('.csv', 'a,b,c\n1,2,3')
>>> params.get('sep')
','

aw.info_dict_validator(compute_info: Callable[[Any], dict], check_info: Callable[[dict], tuple[bool, str]]) → Callable[[Any], tuple[bool, dict]]

Create a validator that computes info then checks it.

This is a two-stage validator: 1. Compute information about the artifact (e.g., df.shape, df.dtypes) 2. Check if that information meets requirements

Parameters:

• compute_info – Function to extract info from artifact

• check_info – Function to check if info is acceptable

Returns:

A validator function

Example

>>> def compute_df_info(df):
...     return {'shape': df.shape, 'null_count': df.isnull().sum().sum()}
>>> def check_df_info(info):
...     if info['null_count'] > 10:
...         return False, "Too many nulls"
...     return True, "OK"
>>> validate = info_dict_validator(compute_df_info, check_df_info)

aw.is_not_empty() → Callable[[Any], tuple[bool, dict]]

Validator that checks artifact is not empty.

Works for sequences, mappings, DataFrames, etc.

aw.is_type(expected_type: type) → Callable[[Any], tuple[bool, dict]]

Validator that checks artifact type.

Example

>>> import pandas as pd
>>> validate = is_type(pd.DataFrame)

aw.load_and_prepare(source_uri: str, target: str = 'generic', validator: Callable = None, max_retries: int = 3) → tuple[Any, dict]

Convenience function to load and prepare data in one call.

Parameters:

• source_uri – URI of data source

• target – Target format/purpose

• validator – Optional custom validator

• max_retries – Maximum retry attempts

Returns:

Tuple of (prepared_dataframe, metadata)

Example

>>> df, meta = load_and_prepare('data.csv', target='cosmo-ready')

aw.load_for_cosmo(source_uri: str, max_retries: int = 3, strict: bool = False) → tuple[Any, dict]

Load and prepare data specifically for cosmograph visualization.

Parameters:

• source_uri – URI of data source

• max_retries – Maximum retry attempts

• strict – If True, actually calls cosmograph for validation

Returns:

Tuple of (prepared_dataframe, metadata)

Example

>>> df, meta = load_for_cosmo('data.csv')
>>> from cosmograph import cosmo
>>> cosmo(df, **meta['preparing']['metadata']['validation_result']['params'])

aw.openai_assistant_from_spec(spec: AgentSpec, *, name: str = None, model: str = 'gpt-4') → dict

Render an OpenAI Assistant-style config dict from an AgentSpec.

The rendering core of to_openai_assistant(), driveable by any AgentSpec.

aw.openai_tools_from_spec(spec: AgentSpec) → list

Render OpenAI function-calling tool schemas from an AgentSpec.

The rendering core of to_openai_tools(), driveable by any AgentSpec.

aw.schema_validator(schema: Any) → Callable[[Any], tuple[bool, dict]]

Create a validator from a schema (Pydantic, JSON Schema, etc.).

Parameters:

schema – A schema object with validation capability

Returns:

A validator function

Example

>>> from pydantic import BaseModel
>>> class DataSchema(BaseModel):
...     x: float
...     y: float
>>> validate = schema_validator(DataSchema)
>>> validate({'x': 1.0, 'y': 2.0})
(True, {'validated': ...})

aw.strict_cosmo_validator() → Callable

Strict validator that actually calls cosmograph.

Requires cosmograph to be installed.

Example

>>> validator = strict_cosmo_validator()
>>> success, info = validator(df)

aw.to_claude_skill(agent: Any, name: str = None, description: str = None, extra_tools: list = None, disable_model_invocation: bool = False) → str

Translate an aw agent to a Claude Code SKILL.md string.

Generates a complete SKILL.md file with YAML frontmatter and markdown instructions, compatible with both Claude Code skills and the Agent Skills Open Standard (agentskills.io).

Parameters:

• agent – An aw agent instance

• name – Override skill name (defaults to kebab-case of class name)

• description – Override description

• extra_tools – Additional Claude Code tools to allow

• disable_model_invocation – If True, skill is manual-only (/name)

Returns:

Complete SKILL.md content as a string

Example

>>> from aw import LoadingAgent
>>> skill = to_claude_skill(LoadingAgent(), name='data-loading')
>>> '---' in skill
True
>>> 'data-loading' in skill
True

aw.to_crewai_yaml(agent: Any, name: str = None, role: str = None, goal: str = None, backstory: str = None) → dict

Translate an aw agent to a CrewAI agent YAML config dict.

CrewAI agents are defined with role, goal, backstory, and tools. This function maps aw’s AgentSpec to that structure.

Parameters:

• agent – An aw agent instance

• name – Override agent name

• role – Override role (defaults to agent description)

• goal – Override goal

• backstory – Override backstory

Returns:

Dict suitable for YAML serialization as a CrewAI agent config

Example

>>> from aw import LoadingAgent
>>> config = to_crewai_yaml(LoadingAgent(), name='data_loader')
>>> config['role']
'Data Loading Specialist'

aw.to_openai_assistant(agent: Any, name: str = None, model: str = 'gpt-4') → dict

Translate an aw agent to an OpenAI Assistant-style config dict.

Generates a configuration suitable for creating an OpenAI Assistant (or Responses API agent) via the API.

Parameters:

• agent – An aw agent instance

• name – Override name

• model – Override model

Returns:

Dict with assistant configuration

Example

>>> from aw import LoadingAgent
>>> config = to_openai_assistant(LoadingAgent())
>>> 'instructions' in config
True

aw.to_openai_tools(agent: Any) → list

Translate an aw agent’s tools to OpenAI function-calling tool schemas.

Generates JSON-Schema-based tool definitions compatible with the OpenAI Chat Completions API and Responses API.

Parameters:

agent – An aw agent instance

Returns:

List of tool definition dicts in OpenAI format

Example

>>> from aw import LoadingAgent
>>> tools = to_openai_tools(LoadingAgent())
>>> all(t['type'] == 'function' for t in tools)
True

aw.try_cosmo_visualization(df: Any, cosmo_function: Callable = None, **cosmo_kwargs) → tuple[bool, dict]

Try to create a cosmograph visualization.

This is a functional validator that actually attempts the visualization.

Parameters:

• df – DataFrame to visualize

• cosmo_function – The cosmograph.cosmo function

• **cosmo_kwargs – Additional arguments for cosmo

Returns:

Tuple of (success, info)

Example

>>> from cosmograph import cosmo
>>> success, info = try_cosmo_visualization(df, cosmo)

aw.workflow_to_crewai_yaml(workflow: Any) → dict

Translate an AgenticWorkflow to CrewAI agents.yaml + tasks.yaml.

Parameters:

workflow – An AgenticWorkflow instance

Returns:

Dict with ‘agents’ and ‘tasks’ keys, each containing YAML-serializable config dicts

Example

>>> from aw import create_cosmo_prep_workflow
>>> workflow = create_cosmo_prep_workflow()
>>> config = workflow_to_crewai_yaml(workflow)
>>> 'agents' in config and 'tasks' in config
True

aw.workflow_to_skills(workflow: Any, output_dir: str | Path) → list

Translate an AgenticWorkflow into a set of Claude Code skills.

Each step in the workflow becomes a separate skill directory.

Parameters:

• workflow – An AgenticWorkflow instance

• output_dir – Parent directory for all skill directories

Returns:

List of Paths to created skill directories

Example

>>> from aw import create_cosmo_prep_workflow
>>> workflow = create_cosmo_prep_workflow()
>>> paths = workflow_to_skills(workflow, '/tmp/skills')

aw.write_skill_directory(agent: Any, output_dir: str | Path, name: str = None, description: str = None, include_scripts: bool = True) → Path

Write a complete Claude Code skill directory.

Creates:

output_dir/ ├── SKILL.md └── scripts/ (if include_scripts and agent has tools)

└── validate.py (validator wrapper script)

Parameters:

• agent – An aw agent instance

• output_dir – Directory to write the skill to

• name – Override skill name

• description – Override description

• include_scripts – Whether to generate helper scripts

Returns:

Path to the created directory

Example

>>> from aw import LoadingAgent
>>> path = write_skill_directory(
...     LoadingAgent(), '/tmp/test-skill', name='data-loading'
... )