Workflow Metrics

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 import Step, Workflow
8from rich.pretty import pprint
9
10hackernews_agent = Agent(
11    name="HackerNews Agent",
12    model=OpenAIResponses(id="gpt-5.2"),
13    tools=[HackerNewsTools()],
14    role="Extract key insights from HackerNews posts",
15)
16
17finance_agent = Agent(
18    name="Finance Agent",
19    model=OpenAIResponses(id="gpt-5.2"),
20    tools=[YFinanceTools()],
21    role="Get stock prices and financial data",
22)
23
24research_team = Team(
25    name="Research Team",
26    members=[hackernews_agent, finance_agent],
27    instructions="Research tech topics from HackerNews and financial data",
28)
29
30content_planner = Agent(
31    name="Content Planner",
32    model=OpenAIResponses(id="gpt-5.2"),
33    instructions="Plan a content schedule based on research",
34)
35
36workflow = Workflow(
37    name="Content Creation Workflow",
38    db=SqliteDb(db_file="tmp/workflow.db"),
39    steps=[
40        Step(name="Research Step", team=research_team),
41        Step(name="Content Planning Step", agent=content_planner),
42    ],
43)
44
45response = workflow.run(input="AI trends in 2024")
46
47# Workflow metrics
48if response.metrics:
49    pprint(response.metrics.to_dict())
50
51# Step metrics
52if response.metrics:
53    for step_name, step_metrics in response.metrics.steps.items():
54        print(f"{step_name}: {step_metrics.executor_name} ({step_metrics.executor_type})")
55        if step_metrics.metrics:
56            print(f"  Duration: {step_metrics.metrics.duration:.2f}s, Tokens: {step_metrics.metrics.total_tokens}")
57
58# Session metrics
59pprint(workflow.get_session_metrics().to_dict())