Compression Events

1"""
2Compression Events
3=============================
4
5Test script to verify compression events are working correctly.
6"""
7
8import asyncio
9
10from kern.agent import Agent
11from kern.models.openai import OpenAIResponses
12from kern.run.agent import RunEvent
13from kern.tools.duckduckgo import DuckDuckGoTools
14
15# ---------------------------------------------------------------------------
16# Create Agent
17# ---------------------------------------------------------------------------
18agent = Agent(
19    model=OpenAIResponses(id="gpt-5-mini"),
20    tools=[DuckDuckGoTools()],
21    description="Specialized in tracking competitor activities",
22    instructions="Use the search tools and always use the latest information and data.",
23    compress_tool_results=True,
24)
25
26
27async def main():
28    print("--- Running agent with compression events ---")
29
30    stream = agent.arun(
31        """
32        Research recent activities for these AI companies:
33        1. OpenAI - latest news
34        2. Anthropic - latest news
35        3. Google DeepMind - latest news
36        """,
37        stream=True,
38        stream_events=True,
39    )
40
41    async for chunk in stream:
42        if chunk.event == RunEvent.run_started.value:
43            print(f"[RunStarted] model={chunk.model}")
44
45        elif chunk.event == RunEvent.model_request_started.value:
46            print(f"[ModelRequestStarted] model={chunk.model}")
47
48        elif chunk.event == RunEvent.model_request_completed.value:
49            print(
50                f"[ModelRequestCompleted] tokens: in={chunk.input_tokens}, out={chunk.output_tokens}"
51            )
52
53        elif chunk.event == RunEvent.tool_call_started.value:
54            print(f"[ToolCallStarted] {chunk.tool.tool_name}")
55
56        elif chunk.event == RunEvent.tool_call_completed.value:
57            print(f"[ToolCallCompleted] {chunk.tool.tool_name}")
58
59        elif chunk.event == RunEvent.compression_started.value:
60            print("[CompressionStarted]")
61
62        elif chunk.event == RunEvent.compression_completed.value:
63            print(
64                f"[CompressionCompleted] compressed={chunk.tool_results_compressed} results"
65            )
66            print(
67                f"  Original: {chunk.original_size} chars -> Compressed: {chunk.compressed_size} chars"
68            )
69            if chunk.original_size and chunk.compressed_size:
70                ratio = (1 - chunk.compressed_size / chunk.original_size) * 100
71                print(f"  Compression ratio: {ratio:.1f}% reduction")
72
73        elif chunk.event == RunEvent.run_completed.value:
74            print("[RunCompleted]")
75
76
77# ---------------------------------------------------------------------------
78# Run Agent
79# ---------------------------------------------------------------------------
80if __name__ == "__main__":
81    asyncio.run(main())

1# Clone and setup repo
2git clone https://github.com/kern-ai/kern.git
3cd kern/cookbook/02_agents/14_advanced
4
5# Create and activate virtual environment
6./scripts/demo_setup.sh
7source .venvs/demo/bin/activate
8
9python compression_events.py