News Scout

The news_scout example (under examples/news_scout/) is a runnable PydanticAI agent that searches news sources, investigates articles, and scores what's worth surfacing. Every model request and every tool call is its own Kitaru checkpoint — individually cached, individually replayable, and visible in the dashboard.

It's a good reference if you want to see:

• KitaruAgent(granular_checkpoints=True) wired end-to-end
• A final @checkpoint that promotes the agent's free-text output to a named artifact (final_report) readable straight from the dashboard
• Namespace memory read detached (outside the flow) so granular-mode agents see concrete values, not artifact refs
• ImageSettings.secret_environment_from for provider API keys, injected automatically when the active stack is remote

What you get

@flow news_scout(interests)
  ├── scout_agent.run_sync(prompt)        ← runs at flow scope
  │     ├── @checkpoint: model_request_1
  │     ├── @checkpoint: tool_call search_news
  │     ├── @checkpoint: model_request_2
  │     ├── @checkpoint: tool_call investigate
  │     ├── @checkpoint: model_request_3
  │     └── …
  └── @checkpoint: publish_report         ← produces `final_report` artifact

A typical sweep takes 2–3 minutes, makes ~10 model requests, and ~20 tool calls, each one its own checkpoint.

Install and run

git clone https://github.com/zenml-io/kitaru
cd kitaru/examples/news_scout

uv sync --extra local --extra pydantic-ai --extra llm
kitaru init
kitaru login        # local dashboard at http://127.0.0.1:8383

ANTHROPIC_API_KEY=sk-ant-...
XAI_API_KEY=xai-...         # optional — enables the search_twitter tool

python scout.py --seed-profile
python scout.py

python scout.py --interests "robotics,biotech"

Why the agent runs at flow scope

Granular mode can't coexist with an enclosing @checkpoint — the adapter falls back to inline execution inside a parent checkpoint, which hides the per-call granularity. To keep each tool call as its own durable step, scout_agent.run_sync(...) lives directly in the flow body.

That constraint has a knock-on effect: agent inputs must be concrete Python values, not artifact refs from kitaru.memory.get(). So the example reads memory outside the flow and passes it in as a flow argument:

def main() -> int:
    load_dotenv()
    # ... parse args ...

    # Detached memory reads — namespace scope supports this
    memory.configure(scope="news_scout", scope_type="namespace")
    interests_from_memory = memory.get("interests")
    interests = override or interests_from_memory or DEFAULT_INTERESTS

    news_scout.run(interests=interests)

The final_report artifact

granular_checkpoints=True captures every internal model and tool call as its own artifact. That's great for replay but it means a reader has to scroll through dozens of entries to find the agent's final answer. The publish_report checkpoint fixes that:

@checkpoint
def publish_report(report_text: str) -> Annotated[str, "final_report"]:
    """Promote the agent's output to a named artifact on the flow."""
    print(report_text)
    return report_text

@flow(image=SCOUT_IMAGE)
def news_scout(interests: list[str]) -> str:
    result = scout_agent.run_sync(build_user_prompt(interests), usage_limits=...)
    return publish_report(report_text=result.output)

In the dashboard, the flow now has a final_report artifact you can open without navigating the tool trace. The per-call checkpoints are still there if you want to replay one of them.

Running on a remote stack

The flow declares its own image; main() attaches a runtime secret reference automatically when the active stack is remote (Kubernetes, Vertex, SageMaker, or AzureML):

SCOUT_IMAGE = ImageSettings(
    requirements=["pydantic-ai-slim[anthropic,openai]>=1.75,<1.80"],
    environment=_collect_non_secret_env(),    # KITARU_SCOUT_MODEL, KITARU_GROK_MODEL
)

@flow(image=SCOUT_IMAGE)
def news_scout(interests: list[str]) -> str:
    ...

At .run() time, _image_override_for_active_stack() checks classify_stack_deployment_type() and, if the stack is remote, attaches ImageSettings.secret_environment_from=["news-scout-keys"] to the run. Kitaru forwards that list to ZenML via Pipeline.with_options(secrets=[...]), so each key in the secret is resolved at step dispatch time and exposed as an environment variable inside the pod. Values never enter image layers, logs, or the frozen execution spec — only the secret name is persisted.

Create the secret once (shared across stacks):

kitaru secrets set news-scout-keys \
  --ANTHROPIC_API_KEY=sk-ant-... \
  --XAI_API_KEY=xai-...      # optional, unlocks search_twitter

kitaru stack use my-k8s-stack
python scout.py

Swapping models

The default model is anthropic:claude-sonnet-4-6. Any PydanticAI model string works:

KITARU_SCOUT_MODEL=openai:gpt-4o python scout.py
KITARU_SCOUT_MODEL=gemini:gemini-2.5-flash python scout.py
KITARU_SCOUT_MODEL=ollama:llama3.3 python scout.py

The Grok model used by the search_twitter tool is overridable separately via KITARU_GROK_MODEL.

Replay a failed step

Because every tool call is a checkpoint, you can replay from the specific call that failed without re-running the upstream work:

kitaru executions list
kitaru executions replay <exec_id> --from search_news_tool_3
kitaru executions retry <exec_id>

The agent config also retries automatically: model_checkpoint_config={"retries": 2} and tool_checkpoint_config={"retries": 1}.

Related