Class-based Executor

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
42# Define executor using a class.
43# Make sure you define the `__call__` method that receives
44# the `step_input`.
45class CustomContentPlanning:
46    def __call__(self, step_input: StepInput) -> StepOutput:
47        """
48        Custom function that does intelligent content planning with context awareness
49        """
50        message = step_input.input
51        previous_step_content = step_input.previous_step_content
52
53        # Create intelligent planning prompt
54        planning_prompt = f"""
55            STRATEGIC CONTENT PLANNING REQUEST:
56            Core Topic: {message}
57            Research Results: {previous_step_content[:500] if previous_step_content else "No research results"}
58            Planning Requirements:
59            1. Create a comprehensive content strategy based on the research
60            2. Leverage the research findings effectively
61            3. Identify content formats and channels
62            4. Provide timeline and priority recommendations
63            5. Include engagement and distribution strategies
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                **Planning Topic:** {message}
73                **Research Integration:** {"Research-based" if previous_step_content else "No research foundation"}
74                **Content Strategy:**
75                {response.content}
76                **Custom Planning Enhancements:**
77                - Research Integration: {"High" if previous_step_content else "Baseline"}
78                - Strategic Alignment: Optimized for multi-channel distribution
79                - Execution Ready: Detailed action items included
80            """.strip()
81
82            return StepOutput(content=enhanced_content)
83
84        except Exception as e:
85            return StepOutput(
86                content=f"Custom content planning failed: {str(e)}",
87                success=False,
88            )
89
90
91# Define steps using different executor types
92
93research_step = Step(
94    name="Research Step",
95    team=research_team,
96)
97
98content_planning_step = Step(
99    name="Content Planning Step",
100    executor=CustomContentPlanning(),
101)
102
103
104# Define and use examples
105if __name__ == "__main__":
106    content_creation_workflow = Workflow(
107        name="Content Creation Workflow",
108        description="Automated content creation with custom execution options",
109        db=SqliteDb(
110            session_table="workflow_session",
111            db_file="tmp/workflow.db",
112        ),
113        # Define the sequence of steps
114        # First run the research_step, then the content_planning_step
115        # You can mix and match agents, teams, and even regular python functions directly as steps
116        steps=[research_step, content_planning_step],
117    )
118    content_creation_workflow.print_response(
119        input="AI trends in 2024",
120        markdown=True,
121    )
122
123    print("\n" + "=" * 60 + "\n")