Step with function

This example demonstrates how to use named steps with custom function executors.

This example demonstrates Workflows using named Step objects with custom function executors. This pattern combines the benefits of named steps with the flexibility of custom functions, allowing for sophisticated data processing within structured workflow steps.

When to use: When you need named step organization but want custom logic that goes beyond what agents/teams provide. Ideal for complex data processing, multi-step operations, or when you need to orchestrate multiple agents within a single step.

1from kern.agent import Agent
2from kern.db.sqlite import SqliteDb
3from kern.models.openai import OpenAIResponses
4from kern.team import Team
5from kern.tools.hackernews import HackerNewsTools
6from kern.tools.yfinance import YFinanceTools
7from kern.workflow.step import Step, StepInput, StepOutput
8from kern.workflow.workflow import Workflow
9
10# Define agents
11hackernews_agent = Agent(
12    name="Hackernews Agent",
13    model=OpenAIResponses(id="gpt-5.2"),
14    tools=[HackerNewsTools()],
15    instructions="Extract key insights and content from Hackernews posts",
16)
17
18finance_agent = Agent(
19    name="Finance Agent",
20    model=OpenAIResponses(id="gpt-5.2"),
21    tools=[YFinanceTools()],
22    instructions="Get stock prices and financial data",
23)
24
25# Define research team for complex analysis
26research_team = Team(
27    name="Research Team",
28    members=[hackernews_agent, finance_agent],
29    instructions="Analyze content and create comprehensive social media strategy",
30)
31
32content_planner = Agent(
33    name="Content Planner",
34    model=OpenAIResponses(id="gpt-5.2"),
35    instructions=[
36        "Plan a content schedule over 4 weeks for the provided topic and research content",
37        "Ensure that I have posts for 3 posts per week",
38    ],
39)
40
41
42def custom_content_planning_function(step_input: StepInput) -> StepOutput:
43    """
44    Custom function that does intelligent content planning with context awareness
45    """
46    message = step_input.input
47    previous_step_content = step_input.previous_step_content
48
49    # Create intelligent planning prompt
50    planning_prompt = f"""
51        STRATEGIC CONTENT PLANNING REQUEST:
52
53        Core Topic: {message}
54
55        Research Results: {previous_step_content[:500] if previous_step_content else "No research results"}
56
57        Planning Requirements:
58        1. Create a comprehensive content strategy based on the research
59        2. Leverage the research findings effectively
60        3. Identify content formats and channels
61        4. Provide timeline and priority recommendations
62        5. Include engagement and distribution strategies
63
64        Please create a detailed, actionable content plan.
65    """
66
67    try:
68        response = content_planner.run(planning_prompt)
69
70        enhanced_content = f"""
71            ## Strategic Content Plan
72
73            **Planning Topic:** {message}
74
75            **Research Integration:** {"Research-based" if previous_step_content else "No research foundation"}
76
77            **Content Strategy:**
78            {response.content}
79
80            **Custom Planning Enhancements:**
81            - Research Integration: {"High" if previous_step_content else "Baseline"}
82            - Strategic Alignment: Optimized for multi-channel distribution
83            - Execution Ready: Detailed action items included
84        """.strip()
85
86        return StepOutput(content=enhanced_content)
87
88    except Exception as e:
89        return StepOutput(
90            content=f"Custom content planning failed: {str(e)}",
91            success=False,
92        )
93
94
95# Define steps using different executor types
96
97research_step = Step(
98    name="Research Step",
99    team=research_team,
100)
101
102content_planning_step = Step(
103    name="Content Planning Step",
104    executor=custom_content_planning_function,
105)
106
107
108# Define and use examples
109if __name__ == "__main__":
110    content_creation_workflow = Workflow(
111        name="Content Creation Workflow",
112        description="Automated content creation with custom execution options",
113        db=SqliteDb(
114            session_table="workflow_session",
115            db_file="tmp/workflow.db",
116        ),
117        # Define the sequence of steps
118        # First run the research_step, then the content_planning_step
119        # You can mix and match agents, teams, and even regular python functions directly as steps
120        steps=[research_step, content_planning_step],
121    )
122    content_creation_workflow.print_response(
123        input="AI trends in 2024",
124        markdown=True,
125    )
126
127    print("\n" + "=" * 60 + "\n")