#4654 Skills & Pydantic Classes Libraries and Python Workflows

#4654 Can python imports use variable names for the library to be imported?

Not directly with the standard import statement, but yes, you can do it dynamically!

Standard Python syntax like import my_variable will try to import a module literally named "my_variable". To use a string stored in a variable, you have a few clean options depending on your use case.

1. The Recommended Way: `importlib.import_module()`

The official standard library approach is using the importlib module. It's clean, safe, and readable.

import importlib

# The library name stored in a variable
module_name = "math"

# Dynamically import the module
math_lib = importlib.import_module(module_name)

# Use it just like a regular imported module
print(math_lib.sqrt(16))  # Output: 4.0

If you need a specific function or class from that module, you can pair it with getattr():

# Equivalent to: from math import sqrt
sqrt_func = getattr(math_lib, "sqrt")
print(sqrt_func(25))  # Output: 5.0

2. The Built-in Way: `import()`

Python has a built-in __import__() function, though the official documentation generally recommends importlib because __import__() can be a bit awkward to use with submodules.

module_name = "random"

# Import using the built-in function
random_lib = __import__(module_name)

print(random_lib.randint(1, 10))

Comparison Summary

Method	Pros	Cons
importlib.import_module()	Standard, clean, handles nested packages smoothly (e.g., `"os.path"`).	Requires importing `importlib` first.
__import__()	Built-in, requires no external imports.	Handles submodules unintuitively; less readable.

A Quick Warning on Safety: Avoid passing untrusted user input directly into dynamic imports, as it can allow users to execute arbitrary code on your system.

Could I use this to import classes for pydantic workflows using a class name identified in an identified skill?

Yes, absolutely. You can use standard dynamic imports (importlib) to dynamically instantiate Pydantic models, agent tools, or workflow steps based on string metadata provided by a skill or config file.

This pattern is common in AI agent frameworks (like Pydantic AI) when building plugin systems, routing nodes in execution graphs, or dynamically setting structured output models.

Basic Pattern: Loading a Pydantic Model from Skill Metadata

Say a "skill" configuration specifies which output schema or payload model it needs to process data:

# Config or Skill definition payload (e.g., loaded from YAML/JSON)
skill_config = {
    "skill_name": "user_extractor",
    "module_path": "my_app.schemas",
    "class_name": "UserProfile",
}

You can resolve the module and fetch the Pydantic BaseModel class like this:

import importlib
from pydantic import BaseModel


def load_pydantic_class(module_path: str, class_name: str) -> type[BaseModel]:
    # 1. Dynamically import the Python module
    module = importlib.import_module(module_path)

    # 2. Extract the class from the module
    cls = getattr(module, class_name)

    # Optional: Validate that it's actually a Pydantic model
    if not (isinstance(cls, type) and issubclass(cls, BaseModel)):
        raise TypeError(f"{class_name} is not a valid Pydantic BaseModel")

    return cls


# Usage
ModelClass = load_pydantic_class(
    skill_config["module_path"], skill_config["class_name"]
)

# Instantiate or validate data with the dynamically loaded model
data = {"username": "alex99", "email": "alex@example.com"}
user_instance = ModelClass(**data)
print(user_instance)

Example: Dynamic Structured Output in Pydantic AI

If you are using dynamic class resolution with an agent framework like Pydantic AI, you can pass dynamically imported models into the agent's expected output schema:

import importlib
from pydantic_ai import Agent

# Suppose your skill defines which response model to use for an LLM response
skill_definition = {
    "model_module": "my_app.schemas",
    "model_class": "FinancialReport",
}

# Dynamically import the model class
ReportModel = getattr(
    importlib.import_module(skill_definition["model_module"]),
    skill_definition["model_class"],
)

# Pass the dynamic model as the result type to the agent
agent = Agent(
    "openai:gpt-4o",
    result_type=ReportModel,  # Dynamic Pydantic schema enforcing LLM response structure
    system_prompt="Analyze the input data and generate the report.",
)

# Run the agent
result = agent.run_sync("Quarterly revenue grew by 15%...")
# result.data will be an instance of financial report

Best Practices & Things to Watch Out For

Fully Qualified Strings: If your skill definition passes a single dot-delimited string (e.g., "my_app.schemas.UserProfile"), you can split it programmatically:

full_path = "my_app.schemas.UserProfile"
module_path, class_name = full_path.rsplit(".", 1)
cls = getattr(importlib.import_module(module_path), class_name)

Type Checking: Python's static type checkers (like mypy or pyright) won't know the specific schema of ModelClass at write-time because it is evaluated at runtime.
Module Registry (Safer Alternative): Rather than letting skill configs import arbitrary modules across your codebase, consider creating an explicit registry mapping skill string keys to Pydantic classes:
# Explicit mapping reduces security/runtime risks SKILL_MODELS = { "user_extractor": UserProfileModel, "sentiment_analyzer": SentimentModel, }

Say I have a skills library and a classes library and specify a workflow of skills

This is a classic architecture for building dynamic, configurable workflows (like AI agents, ETL pipelines, or task orchestration engines).

By decoupling Skills (the step logic) from Classes/Schemas (the payload data models), you can construct workflows purely from dynamic configurations (like JSON or YAML files).

Here is a complete, runnable example of how to orchestrate this clean dependency-injection pipeline using importlib.

Step 1: Project Structure Example

Suppose your project is organized like this:

my_project/
├── classes/
├──   __init__.py
├──   schemas.py        # Contains your Pydantic data models
├── skills/
├──   __init__.py
├──   text_skills.py    # Contains your skill functions/classes
└── runner.py           # The dynamic workflow runner

Step 2: Define Your Data Schemas (`classes/schemas.py`)

from pydantic import BaseModel, Field


class TextPayload(BaseModel):
    raw_text: str


class ProcessedTextPayload(BaseModel):
    processed_text: str
    word_count: int

Step 3: Define Your Skill Functions (`skills/text_skills.py`)

Skills receive a dynamically instantiated Pydantic object and return another Pydantic object.

from my_project.classes.schemas import ProcessedTextPayload, TextPayload


def clean_and_count_skill(payload: TextPayload) -> ProcessedTextPayload:
    cleaned = payload.raw_text.strip().upper()
    count = len(cleaned.split())

    return ProcessedTextPayload(processed_text=cleaned, word_count=count)

Step 4: Define the Workflow Configuration

Your workflow can now be defined entirely as metadata (e.g., loaded from a database, API, or YAML file):

workflow_config = [
    {
        "step_name": "clean_text_step",
        "skill": "my_project.skills.text_skills.clean_and_count_skill",
        "input_class": "my_project.classes.schemas.TextPayload",
    }
]

Step 5: The Orchestrator (`runner.py`)

Here is how you dynamically resolve both the Skill function and the Input Schema, instantiate the input model, and execute the skill step:

import importlib
from typing import Any
from pydantic import BaseModel


def resolve_object(target_string: str) -> Any:
    """Helper to split module path and object name, then dynamically import it.

    e.g., 'my_project.classes.schemas.TextPayload' -> returns TextPayload class
    """
    module_path, obj_name = target_string.rsplit(".", 1)
    module = importlib.import_module(module_path)
    return getattr(module, obj_name)


def run_workflow(config: list[dict], initial_data: dict):
    current_data = initial_data

    for step in config:
        print(f"--- Running Step: {step['step_name']} ---")

        # 1. Dynamically resolve skill function and input Pydantic class
        skill_func = resolve_object(step["skill"])
        input_class: type[BaseModel] = resolve_object(step["input_class"])

        # 2. Validate input data using the dynamically imported Pydantic class
        validated_input = input_class(**current_data)

        # 3. Execute the skill with the Pydantic instance
        step_result = skill_func(validated_input)

        # Print output and update payload for the next step in the pipeline
        print(f"Output type: {type(step_result).__name__}")
        print(f"Output data: {step_result.model_dump()}")

        # If chaining steps, convert back to dict for the next input
        current_data = step_result.model_dump()


# --- Test Run ---
if __name__ == "__main__":
    input_data = {"raw_text": "  hello world from dynamic workflow!  "}
    run_workflow(workflow_config, input_data)

Output

--- Running Step: clean_text_step ---
Output type: ProcessedTextPayload
Output data: {'processed_text': 'HELLO WORLD FROM DYNAMIC WORKFLOW!', 'word_count': 5}

Key Advantages of This Design

Zero Hardcoding: You can add 50 new skills or models without ever modifying the run_workflow executor loop.
Strict Runtime Type Safety: Pydantic will instantly raise a ValidationError if the raw incoming payload doesn't fit the expected input_class schema for that skill step.
Serialization Ready: Workflows can be easily saved, versioned, or loaded from JSON databases or frontend dashboard builders.

Provide a python workflow for "Find, Mine, Refine and Distribute"

Here is a complete Python workflow implementing the Find $\rightarrow$ Mine $\rightarrow$ Refine $\rightarrow$ Distribute pattern.

In this implementation:

Find: Discovers raw targets/URLs/sources.
Mine: Extracts raw, unstructured content from those sources.
Refine: Uses Pydantic to validate, structure, and enrich the mined data into clean output schemas.
Distribute: Routes the refined data to target outputs (e.g., databases, API endpoints, or file stores).

It uses standard importlib dynamic resolving so you can swap out any step's implementation or schema on the fly.

Step 1: Schemas (`classes/schemas.py`)

Define the Pydantic data models for the intermediate and final outputs of your workflow.

from pydantic import BaseModel, HttpUrl, Field


# Initial discovery output
class DiscoveredTarget(BaseModel):
    source_id: str
    url: str


# Raw mined data output
class RawMinedData(BaseModel):
    source_id: str
    raw_text: str


# Refined structured payload
class RefinedInsight(BaseModel):
    source_id: str
    summary: str
    key_topics: list[str]
    sentiment_score: float = Field(..., ge=-1.0, le=1.0)


# Distribution result
class DistributionReceipt(BaseModel):
    source_id: str
    destination: str
    status: str

Step 2: The Skill Library (`skills/pipeline_skills.py`)

Each stage in your pipeline is implemented as a modular skill function.

from my_project.classes.schemas import (
    DiscoveredTarget,
    RawMinedData,
    RefinedInsight,
    DistributionReceipt,
)


def find_sources(config: dict) -> list[DiscoveredTarget]:
    """FIND: Locates raw target resources to process."""
    print("🔍 [FIND] Discovering sources...")
    return [
        DiscoveredTarget(
            source_id="doc_101", url="https://example.com/reports/q3"
        ),
        DiscoveredTarget(
            source_id="doc_102", url="https://example.com/reports/q4"
        ),
    ]


def mine_content(target: DiscoveredTarget) -> RawMinedData:
    """MINE: Pulls unrefined data out of a discovered target."""
    print(f"⛏️  [MINE] Extracting raw data from: {target.url}")
    # Simulated content extraction (e.g., web scraping or PDF parsing)
    raw_content = f"Quarterly growth increased by 12% for {target.source_id}. Market adoption is high."
    return RawMinedData(source_id=target.source_id, raw_text=raw_content)


def refine_data(mined: RawMinedData) -> RefinedInsight:
    """REFINE: Cleans, structures, and enriches raw data using Pydantic."""
    print(f"🧹 [REFINE] Structuring and validating content for: {mined.source_id}")
    # Simulated extraction/LLM processing into structured Pydantic format
    return RefinedInsight(
        source_id=mined.source_id,
        summary=mined.raw_text,
        key_topics=["growth", "market adoption"],
        sentiment_score=0.85,
    )


def distribute_insight(insight: RefinedInsight) -> DistributionReceipt:
    """DISTRIBUTE: Delivers the refined payload to its final destination."""
    print(f"🚀 [DISTRIBUTE] Pushing {insight.source_id} insight to database...")
    # Simulated distribution (e.g., REST API POST, DB write, or message queue)
    return DistributionReceipt(
        source_id=insight.source_id,
        destination="PostgreSQL::analytics.insights",
        status="SUCCESS",
    )

Step 3: Workflow Definition & Dynamic Runner (`runner.py`)

Using dynamic module loading, you can specify your stages via metadata strings.

import importlib
from typing import Any


# Configuration mapping out the Find -> Mine -> Refine -> Distribute pipeline
WORKFLOW_SPEC = {
    "find": {
        "skill": "my_project.skills.pipeline_skills.find_sources",
    },
    "mine": {
        "skill": "my_project.skills.pipeline_skills.mine_content",
        "input_class": "my_project.classes.schemas.DiscoveredTarget",
    },
    "refine": {
        "skill": "my_project.skills.pipeline_skills.refine_data",
        "input_class": "my_project.classes.schemas.RawMinedData",
    },
    "distribute": {
        "skill": "my_project.skills.pipeline_skills.distribute_insight",
        "input_class": "my_project.classes.schemas.RefinedInsight",
    },
}


def resolve_symbol(path_str: str) -> Any:
    """Dynamically loads a function or class from a string path."""
    module_path, name = path_str.rsplit(".", 1)
    module = importlib.import_module(module_path)
    return getattr(module, name)


def execute_pipeline(spec: dict):
    # 1. FIND STAGE
    find_func = resolve_symbol(spec["find"]["skill"])
    discovered_targets = find_func({})

    receipts = []

    # Process each discovered target through Mine -> Refine -> Distribute
    for target in discovered_targets:
        print(f"\n--- Processing Target: {target.source_id} ---")

        # 2. MINE STAGE
        mine_func = resolve_symbol(spec["mine"]["skill"])
        mine_class = resolve_symbol(spec["mine"]["input_class"])
        mined_data = mine_func(mine_class(**target.model_dump()))

        # 3. REFINE STAGE
        refine_func = resolve_symbol(spec["refine"]["skill"])
        refine_class = resolve_symbol(spec["refine"]["input_class"])
        refined_insight = refine_func(refine_class(**mined_data.model_dump()))

        # 4. DISTRIBUTE STAGE
        distribute_func = resolve_symbol(spec["distribute"]["skill"])
        distribute_class = resolve_symbol(spec["distribute"]["input_class"])
        receipt = distribute_func(distribute_class(**refined_insight.model_dump()))

        receipts.append(receipt)

    print("\n--- Pipeline Complete ---")
    for r in receipts:
        print(f"Receipt: {r.model_dump()}")


if __name__ == "__main__":
    execute_pipeline(WORKFLOW_SPEC)

Output Execution

🔍 [FIND] Discovering sources...

--- Processing Target: doc_101 ---
⛏️  [MINE] Extracting raw data from: https://example.com/reports/q3
🧹 [REFINE] Structuring and validating content for: doc_101
🚀 [DISTRIBUTE] Pushing doc_101 insight to database...

--- Processing Target: doc_102 ---
⛏️  [MINE] Extracting raw data from: https://example.com/reports/q4
🧹 [REFINE] Structuring and validating content for: doc_102
🚀 [DISTRIBUTE] Pushing doc_102 insight to database...

--- Pipeline Complete ---
Receipt: {'source_id': 'doc_101', 'destination': 'PostgreSQL::analytics.insights', 'status': 'SUCCESS'}
Receipt: {'source_id': 'doc_102', 'destination': 'PostgreSQL::analytics.insights', 'status': 'SUCCESS'}

From <https://gemini.google.com/app/dfc35497df2ef774> Google Gemini (3.6 Flash)

Search This Blog

Learning from ChatGPT