SKILL.md
---
name: developing-genkit-python
description: Develop AI-powered applications using Genkit in Python. Use when the user asks about Genkit, AI agents, flows, or tools in Python, or when encountering Genkit errors, import issues, or API problems.
metadata:
genkit-managed: true
---
# Genkit Python
## Prerequisites
- **Runtime**: Python **3.14+**, **`uv`** for deps ([install](https://docs.astral.sh/uv/getting-started/installation/)).
- **CLI**: `genkit --version` — install via `npm install -g genkit-cli` if missing.
**New projects:** [Setup](references/setup.md) (bootstrap + env). **Patterns and code samples:** [Examples](references/examples.md).
## Hello World
```python
from genkit import Genkit
from genkit.plugins.google_genai import GoogleAI
ai = Genkit(
plugins=[GoogleAI()],
model='googleai/gemini-flash-latest',
)
async def main():
response = await ai.generate(prompt='Tell me a joke about Python.')
print(response.text)
if __name__ == '__main__':
ai.run_main(main())
```
## Critical: Do Not Trust Internal Knowledge
The Python SDK changes often — verify imports and APIs against the references here or upstream docs. On **any** error, read [Common Errors](references/common-errors.md) first.
## Development Workflow
1. Default provider: **Google AI** (`GoogleAI()`), **`GEMINI_API_KEY`** in the environment.
2. Model IDs: always prefixed, e.g. **`googleai/gemini-flash-latest`** (always-on-latest Flash alias; same pattern as other skills).
3. Entrypoint: **`ai.run_main(main())`** for Genkit-driven apps (not `asyncio.run()` for long-lived servers started with `genkit start` — see [Common Errors](references/common-errors.md)).
4. After generating code, follow [Dev Workflow](references/dev-workflow.md) for `genkit start` and the Dev UI.
5. On errors: step 1 is always [Common Errors](references/common-errors.md).
## References
- [Examples](references/examples.md): Structured output, streaming, flows, tools, embeddings.
- [Setup](references/setup.md): New project bootstrap and plugins.
- [Common Errors](references/common-errors.md): Read first when something breaks.
- [FastAPI](references/fastapi.md): HTTP, `genkit_fastapi_handler`, parallel flows.
- [Dotprompt](references/dotprompt.md): `.prompt` files and helpers.
- [Evals](references/evals.md): Evaluators and datasets.
- [Dev Workflow](references/dev-workflow.md): `genkit start`, Dev UI, checklist.
references/common-errors.md
# Common Errors — Genkit Python
## Before anything else: read this file when you hit any error.
---
## ModuleNotFoundError: No module named 'genkit.plugins.google_genai'
**Cause:** Plugin package not installed.
**Fix:** Add dependencies from PyPI:
```bash
uv add genkit genkit-plugin-google-genai
```
---
## 400 INVALID_ARGUMENT: functionDeclaration parameters schema should be of type OBJECT
**Cause:** Tool function has bare scalar parameters (e.g. `city: str`). Gemini requires object schema.
**Fix:** Wrap parameters in a Pydantic BaseModel:
```python
from pydantic import BaseModel
# Wrong
@ai.tool()
async def get_weather(city: str) -> str: ...
# Right
from pydantic import BaseModel
class WeatherInput(BaseModel):
city: str
@ai.tool()
async def get_weather(input: WeatherInput) -> str: ...
```
---
## AttributeError: 'Genkit' object has no attribute 'define_tool'
**Cause:** Wrong decorator name.
**Fix:** Use `@ai.tool()`, not `@ai.define_tool()`.
---
## RuntimeError / event loop errors when using asyncio.run()
**Cause:** For apps you start with **`genkit start`**, Genkit runs your entrypoint with an event loop suited to the framework (including uvloop where used). There is no “default” loop for you to manage in that mode.
**Fix:** For long-running Genkit apps (servers, flows served under `genkit start`), use **`ai.run_main(main())`** as your entrypoint instead of `asyncio.run(main())`. For one-off scripts that exit when done, using `asyncio.run()` can still be appropriate when you are not using `genkit start`.
---
## Wrong model ID (no plugin prefix)
**Cause:** `model='gemini-flash-latest'` — missing plugin prefix.
**Fix:** `model='googleai/gemini-flash-latest'`
---
## response.json / response.message AttributeError
- Use `response.text` for plain text output
- Use `response.output` for structured (JSON) output
---
## await ai.generate_stream(...) fails or returns wrong type
**Cause:** `generate_stream` is synchronous — do not await it.
**Fix:**
```python
sr = ai.generate_stream(prompt='...') # no await
async for chunk in sr.stream: ...
final = await sr.response
```
references/dotprompt.md
# Dotprompt — Genkit Python
## What it is
`.prompt` files combine YAML frontmatter (model config, schemas) with Handlebars templates. Keeps prompt logic out of Python code and makes variants easy.
## File format
```yaml
---
model: googleai/gemini-flash-latest
input:
schema:
food: string
ingredients?(array): string # ? = optional
output:
schema: Recipe # references a schema registered with ai.define_schema()
format: json
---
You are a chef. Generate a recipe for {{food}}.
{{#if ingredients}}
Include these ingredients:
{{list ingredients}}
{{/if}}
```
Place `.prompt` files in a `prompts/` directory and point `prompt_dir` at it.
## Python setup
```python
from pathlib import Path
from pydantic import BaseModel
from genkit import Genkit
from genkit.plugins.google_genai import GoogleAI
ai = Genkit(
plugins=[GoogleAI()],
model='googleai/gemini-flash-latest',
prompt_dir=Path(__file__).resolve().parent.parent / 'prompts',
)
# Register Pydantic models referenced in .prompt output.schema
class Recipe(BaseModel):
title: str
steps: list[str]
ai.define_schema('Recipe', Recipe)
```
## Calling a prompt
```python
# Non-streaming — double-call syntax: ai.prompt('name')(input={...})
response = await ai.prompt('recipe')(input={'food': 'banana bread'})
result = Recipe.model_validate(response.output)
# Variant (recipe.robot.prompt file)
response = await ai.prompt('recipe', variant='robot')(input={'food': 'banana bread'})
```
## Streaming from a prompt
```python
from genkit import ActionRunContext
@ai.flow()
async def tell_story(subject: str, ctx: ActionRunContext) -> str:
result = ai.prompt('story').stream(input={'subject': subject})
full = ''
async for chunk in result.stream:
if chunk.text:
ctx.send_chunk(chunk.text)
full += chunk.text
return full
```
Note: `.stream(input={...})` not `ai.generate_stream(...)` — different call shape for prompts.
## Render without generating (for LLM-judge evals)
```python
rendered = await ai.prompt('my_prompt').render(input={'key': 'value'})
response = await ai.generate(model='googleai/gemini-flash-latest', messages=rendered.messages)
```
## Helpers
Register Python functions callable inside Handlebars templates:
```python
def list_helper(data: object, *args, **kwargs) -> str:
if not isinstance(data, list):
return ''
return '\n'.join(f'- {item}' for item in data)
ai.define_helper('list', list_helper)
```
Then use `{{list ingredients}}` in your `.prompt` file.
## Variants
Name the file `<name>.<variant>.prompt` — e.g. `recipe.robot.prompt`.
Call with `ai.prompt('recipe', variant='robot')`.
## Partials
Use `{{>partial_name param=value}}` in templates. Partial files are named `_partial_name.prompt`.
references/dev-workflow.md
# Dev Workflow — Genkit Python
## Agent responsibility
After generating code, always give the developer:
1. The full pre-run checklist with copy-paste commands using absolute paths
2. The `genkit start` command to run in their terminal (foreground — it's expected to block)
3. Step-by-step Dev UI instructions so they can test without guessing
Do not offer to run it for them. Give them the commands and let them run it.
---
## Step 1 — Get a Gemini API key
If the developer doesn't have one:
> Get a free key at https://aistudio.google.com/apikey — click **"Create API key"**, copy it.
---
## Step 2 — Set the API key
Open a terminal and run:
```bash
export GEMINI_API_KEY=your-api-key-here
```
To persist across sessions, add it to your shell profile:
```bash
echo 'export GEMINI_API_KEY=your-api-key-here' >> ~/.zshrc && source ~/.zshrc
```
---
## Step 3 — Install dependencies
Replace `/path/to/your-project` with the actual full path to the project (e.g. `/Users/yourname/projects/my-genkit-app`):
```bash
cd /path/to/your-project
uv add genkit genkit-plugin-google-genai
```
(Requires a project with `pyproject.toml` — run `uv init` in an empty directory first if needed.)
---
## Step 4 — Start the Dev UI
Run this in your terminal. **It will block — that's expected.** Leave this terminal open while you use the Dev UI.
```bash
cd /path/to/your-project
GEMINI_API_KEY=your-api-key-here genkit start -- uv run src/main.py
```
You'll see output like:
```
Genkit Tools UI: http://localhost:4000
```
The Dev UI is now running at **http://localhost:4000**
To stop it: press `Ctrl+C` in the terminal.
---
## Step 5 — Test in the Dev UI
1. Open **http://localhost:4000** in your browser
2. Click **"Run"** in the left sidebar
3. Find your flow by name (e.g. `summarize`, `chat`, `joke_generator`)
4. In the input box, paste your input as JSON — e.g:
```json
{"text": "hello world"}
```
5. Click the **"Run"** button — the output appears on the right
6. Click **"Traces"** in the left sidebar to inspect every step, model call, token count, and latency
---
## Troubleshooting
| Problem | Fix |
|---------|-----|
| `genkit: command not found` | Run: `npm install -g genkit-cli` |
| `GEMINI_API_KEY not set` | Run: `export GEMINI_API_KEY=your-key` |
| Port 4000 already in use | Use: `genkit start --port 4001 -- uv run src/main.py` |
| `uv: command not found` | Run: `curl -LsSf https://astral.sh/uv/install.sh \| sh` |
| Flow not showing in Dev UI | Make sure `genkit start` output shows no errors |
references/setup.md
# Setup — Genkit Python
## New project
**Always use a virtual environment** — never install Genkit into the system interpreter. With **uv**, the project’s **`.venv`** is created and used by `uv sync` / `uv run` automatically once you add dependencies.
```bash
mkdir my-app && cd my-app
uv init
uv venv --python 3.14 .venv
# Unix: source .venv/bin/activate
# Windows: .venv\Scripts\activate
uv add genkit genkit-plugin-google-genai
export GEMINI_API_KEY=your_key_here
```
`uv init` creates `pyproject.toml`. Add your app under something like `src/main.py` (or match whatever layout `uv` generated) and point `genkit start` at that entrypoint.
## pyproject.toml
Minimal `[project]` block with unpinned Genkit deps (resolver picks compatible releases):
```toml
[project]
name = "my-app"
version = "0.1.0"
requires-python = ">=3.14"
dependencies = [
"genkit",
"genkit-plugin-google-genai",
]
```
## Plugins
Packages are **`genkit-plugin-*`** on PyPI, e.g. `genkit-plugin-google-genai`, `genkit-plugin-vertex-ai`, `genkit-plugin-anthropic`, `genkit-plugin-fastapi`. Install with `uv add genkit-plugin-<name>`.
## Python version
**3.14+**. Always use a `venv` using `uv venv --python 3.14 .venv` when creating the environment before you run any commands.
references/fastapi.md
# FastAPI — Genkit Python
## Install
```bash
uv add genkit-plugin-fastapi fastapi uvicorn
```
---
## Streaming by default
The `genkit_fastapi_handler` decorator auto-streams when the client sends `Accept: text/event-stream`.
No extra setup — just add the header on the frontend and it works.
**Wire format (SSE):**
```
data: {"message": "<chunk text>"} ← one per ctx.send_chunk() call
data: {"message": "<chunk text>"}
data: {"result": <final output>} ← sent once when flow completes
```
**Frontend (JS EventSource):**
```js
const res = await fetch('/flow/chat', {
method: 'POST',
headers: { 'Content-Type': 'application/json', 'Accept': 'text/event-stream' },
body: JSON.stringify({ data: { topic: 'quantum computing' } }),
});
const reader = res.body.getReader();
// decode and parse each `data: {...}` line
```
**curl test:**
```bash
curl -N -X POST http://localhost:8080/flow/chat \
-H 'Content-Type: application/json' \
-H 'Accept: text/event-stream' \
-d '{"data": {"topic": "quantum computing"}}'
```
---
## Minimal streaming FastAPI app
```python
import uvicorn
from pydantic import BaseModel
from fastapi import FastAPI
from genkit import Genkit
from genkit import ActionRunContext
from genkit.plugins.fastapi import genkit_fastapi_handler
from genkit.plugins.google_genai import GoogleAI
ai = Genkit(plugins=[GoogleAI()], model='googleai/gemini-flash-latest')
app = FastAPI()
class ChatInput(BaseModel):
topic: str
@app.post('/flow/chat', response_model=None)
@genkit_fastapi_handler(ai)
@ai.flow()
async def chat(input: ChatInput, ctx: ActionRunContext) -> str:
sr = ai.generate_stream(prompt=f'Tell me about {input.topic}.')
full = ''
async for chunk in sr.stream:
if chunk.text:
ctx.send_chunk(chunk.text) # each chunk → SSE event on the wire
full += chunk.text
return full
if __name__ == '__main__':
uvicorn.run(app, host='0.0.0.0', port=8080)
```
**Key:** flow must accept `ctx: ActionRunContext` and call `ctx.send_chunk(text)` to emit SSE chunks.
Without `ctx.send_chunk`, the flow runs but streams nothing — client waits for the final result.
---
## Advanced Use Cases
### Fine-grained control over flow streaming
Complex apps chain flows — a parent orchestrates children. Chunks propagate upward by **passing `ctx` to child flows**.
```python
class ResearchInput(BaseModel):
topic: str
@ai.flow()
async def research(input: ResearchInput, ctx: ActionRunContext) -> str:
"""Child flow — streams its generate_stream chunks to whoever called it."""
sr = ai.generate_stream(prompt=f'Explain {input.topic} in depth.')
full = ''
async for chunk in sr.stream:
if chunk.text:
ctx.send_chunk(chunk.text) # propagates up through the call stack
full += chunk.text
return full
class HeadlineInput(BaseModel):
text: str
@ai.flow()
async def make_headline(input: HeadlineInput) -> str:
"""Child flow — non-streaming, returns instantly."""
response = await ai.generate(prompt=f'One-line headline for: {input.text}')
return response.text.strip()
class ReportInput(BaseModel):
topic: str
@app.post('/flow/report', response_model=None)
@genkit_fastapi_handler(ai)
@ai.flow()
async def report(input: ReportInput, ctx: ActionRunContext) -> str:
"""Parent flow — calls children, composes a streaming report."""
# Step 1: fast non-streaming call
headline = await make_headline(HeadlineInput(text=input.topic))
ctx.send_chunk(f'# {headline}\n\n') # send headline immediately
# Step 2: child flow streams its chunks — passes ctx so they flow up
body = await research(ResearchInput(topic=input.topic), ctx)
return f'# {headline}\n\n{body}'
```
**Rules for nested streaming:**
- Child flows that should stream must also accept `ctx: ActionRunContext`
- Pass the parent's `ctx` when calling child flows: `await child(input, ctx)`
- Non-streaming child flows don't need `ctx` — just `await` them normally
- A child that doesn't call `ctx.send_chunk` contributes nothing to the stream (fine for parallel data fetching)
### Executing flows in parallel
Use `asyncio.gather` to run multiple flows concurrently. Only makes sense when children don't need to stream.
```python
import asyncio
class AnalysisInput(BaseModel):
text: str
class CheckResult(BaseModel):
issues: list[str]
class CombinedAnalysis(BaseModel):
issues: list[str]
@ai.flow()
async def check_security(input: AnalysisInput) -> CheckResult:
# Here the model reviews the text; replace with your real prompt/schema as needed.
r = await ai.generate(
prompt=f'List security concerns as a short comma-separated line (or "none"): {input.text[:2000]}',
)
raw = (r.text or '').strip()
issues = [s.strip() for s in raw.split(',') if s.strip() and s.strip().lower() != 'none']
return CheckResult(issues=issues)
@ai.flow()
async def check_bugs(input: AnalysisInput) -> CheckResult:
# Model lists possible bugs; tune prompt for your codebase.
r = await ai.generate(
prompt=f'List likely bugs or correctness issues as a short comma-separated line (or "none"): {input.text[:2000]}',
)
raw = (r.text or '').strip()
issues = [s.strip() for s in raw.split(',') if s.strip() and s.strip().lower() != 'none']
return CheckResult(issues=issues)
@ai.flow()
async def check_style(input: AnalysisInput) -> CheckResult:
# Model suggests style/clarity issues; optional: use output_schema for structured rows.
r = await ai.generate(
prompt=f'List style or clarity issues as a short comma-separated line (or "none"): {input.text[:2000]}',
)
raw = (r.text or '').strip()
issues = [s.strip() for s in raw.split(',') if s.strip() and s.strip().lower() != 'none']
return CheckResult(issues=issues)
@app.post('/flow/analyze', response_model=None)
@genkit_fastapi_handler(ai)
@ai.flow()
async def analyze(input: AnalysisInput) -> CombinedAnalysis:
security, bugs, style = await asyncio.gather(
check_security(input),
check_bugs(input),
check_style(input),
)
return CombinedAnalysis(issues=security.issues + bugs.issues + style.issues)
```
---
## Structured output endpoint (non-streaming)
```python
class SentimentResult(BaseModel):
sentiment: str # positive / negative / neutral
confidence: float # 0.0–1.0
key_phrases: list[str]
@app.post('/flow/sentiment', response_model=None)
@genkit_fastapi_handler(ai)
@ai.flow()
async def sentiment(input: AnalysisInput) -> SentimentResult:
response = await ai.generate(
prompt=f'Analyze sentiment: {input.text}',
output_format='json',
output_schema=SentimentResult,
)
return response.output
```
Client calls this without `Accept: text/event-stream` — gets `{"result": {...}}` back.
---
## Decorator order
Must be exactly: `@app.post` → `@genkit_fastapi_handler(ai)` → `@ai.flow()`
```python
@app.post('/flow/chat', response_model=None) # 1. FastAPI route
@genkit_fastapi_handler(ai) # 2. Genkit wire format + streaming
@ai.flow() # 3. Flow registration
async def chat(input: ChatInput, ctx: ActionRunContext) -> str:
...
```
---
## Run with Dev UI
```bash
GEMINI_API_KEY=your-key genkit start -- uv run src/main.py
```
Leave the process running until the CLI prints something like:
```
Genkit Developer UI: http://localhost:4000
```
Open that URL. Port may differ if 4000 is busy.references/examples.md
# Genkit Python Examples
Minimal patterns for common Genkit APIs. Examples use **Google AI** (`GoogleAI`, `googleai/...`); other providers use the same patterns with the right plugin and model prefix.
## Public imports
Use **`genkit`**, **`genkit.plugins.*`**, **`genkit.embedder`**, **`genkit.evaluator`**, and **`genkit.model`** (and similar public modules) only — not internal packages (`genkit._core`, etc.).
```python
from genkit import Genkit, ActionRunContext
from genkit.plugins.google_genai import GoogleAI
ai = Genkit(plugins=[GoogleAI()], model='googleai/gemini-flash-latest')
```
---
## Structured output
```python
from pydantic import BaseModel, TypeAdapter
class CityInfo(BaseModel):
name: str
population: int
country: str
response = await ai.generate(
prompt='Give facts about Tokyo.',
output_format='json',
output_schema=CityInfo,
)
city = response.output
# Arrays
schema = TypeAdapter(list[CityInfo]).json_schema()
response = await ai.generate(
prompt='List 3 cities.',
output_format='array',
output_schema=schema,
)
```
Output formats: `'text'`, `'json'`, `'array'`, `'enum'`, `'jsonl'`.
---
## Streaming (text)
```python
sr = ai.generate_stream(prompt='Tell me a story.')
async for chunk in sr.stream:
if chunk.text:
print(chunk.text, end='', flush=True)
final = await sr.response # final.text
```
---
## Text and media parts
```python
# Non-streaming
response = await ai.generate(prompt='...')
for media in response.media:
print(media.content_type, (media.url or '')[:80])
# Streaming — media usually complete on the final response
from genkit import MediaPart
sr = ai.generate_stream(prompt='...')
async for chunk in sr.stream:
if chunk.text:
print(chunk.text, end='', flush=True)
final = await sr.response
for media in final.media:
print(media.content_type, (media.url or '')[:80])
if final.message:
for part in final.message.content:
if isinstance(part.root, MediaPart) and part.root.media:
print(part.root.media.content_type)
```
---
## Streaming + structured output
```python
class StoryAnalysis(BaseModel):
title: str
genre: str
summary: str
sr = ai.generate_stream(
prompt='Write a short story then analyze it.',
output_format='json',
output_schema=StoryAnalysis,
)
async for chunk in sr.stream:
if chunk.text:
print(chunk.text, end='', flush=True)
final = await sr.response
analysis = final.output
```
---
## Flows
```python
class SummarizeInput(BaseModel):
text: str
@ai.flow()
async def summarize(input: SummarizeInput) -> str:
response = await ai.generate(prompt=f'Summarize: {input.text}')
return response.text
```
---
## Streaming flows
```python
@ai.flow()
async def stream_story(subject: str, ctx: ActionRunContext) -> str:
sr = ai.generate_stream(prompt=f'Story about {subject}.')
full = ''
async for chunk in sr.stream:
if chunk.text:
ctx.send_chunk(chunk.text)
full += chunk.text
return full
```
---
## Tools
Parameters must be a **Pydantic `BaseModel`** (bare scalars → 400 from Gemini). Use **`@ai.tool()`**, not `@ai.define_tool()`.
```python
class WeatherInput(BaseModel):
city: str
@ai.tool()
async def get_weather(input: WeatherInput) -> str:
return f'Sunny in {input.city}'
response = await ai.generate(prompt='Weather in Paris?', tools=[get_weather])
```
---
## Embeddings
```python
from genkit.plugins.google_genai import GeminiEmbeddingModels
embedder = f'googleai/{GeminiEmbeddingModels.GEMINI_EMBEDDING_001}'
embeddings = await ai.embed(embedder=embedder, content='The sky is blue.')
vector = embeddings[0].embedding
embeddings = await ai.embed_many(
embedder=embedder,
content=['The sky is blue.', 'Grass is green.'],
)
```
Common embedders: `googleai/gemini-embedding-001`, `googleai/gemini-embedding-exp-03-07`.
references/evals.md
# Evals — Genkit Python
## Two types of evaluators
1. **Built-in** — ship with `genkit-plugin-evaluators`, register with `register_genkit_evaluators(ai)`
2. **BYO (LLM-based)** — define your own scoring logic with `ai.define_evaluator()`
## Install
```bash
uv add genkit-plugin-evaluators
```
## Dataset format
A JSON file, one object per test case:
```json
[
{"testCaseId": "case1", "input": "x", "output": "banana", "reference": "ba?a?a"},
{"testCaseId": "case2", "input": "x", "output": "apple", "reference": "ba?a?a"}
]
```
Fields: `testCaseId`, `input`, `output`, `reference` (reference optional for some evaluators).
## Built-in evaluators
```python
from genkit.plugins.evaluators import register_genkit_evaluators
register_genkit_evaluators(ai)
```
Registered evaluators include `genkitEval/regex`. Run via CLI:
```bash
genkit eval:run datasets/my_dataset.json --evaluators=genkitEval/regex
```
## BYO evaluator
```python
from genkit.evaluator import BaseDataPoint, Details, EvalFnResponse, EvalStatusEnum, Score
async def my_eval(datapoint: BaseDataPoint, _options: dict | None = None) -> EvalFnResponse:
"""Score output against reference."""
output = str(datapoint.output or '')
reference = str(datapoint.reference or '')
passed = output.strip() == reference.strip()
return EvalFnResponse(
test_case_id=datapoint.test_case_id or '',
evaluation=Score(
score=1.0 if passed else 0.0,
status=EvalStatusEnum.PASS if passed else EvalStatusEnum.FAIL,
details=Details(reasoning='Exact match check'),
),
)
ai.define_evaluator(
name='byo/my_eval',
display_name='My Eval',
definition='Checks exact match of output vs reference.',
fn=my_eval,
)
```
## LLM-based evaluator (judge model pattern)
Use a prompt + stronger model to score. See `py/samples/evaluators/src/main.py` for full examples (`byo/maliciousness`, `byo/answer_accuracy`).
Core pattern:
```python
async def llm_eval(datapoint: BaseDataPoint, _options: dict | None = None) -> EvalFnResponse:
prompt = ai.prompt('my_judge_prompt')
rendered = await prompt.render(input={'output': str(datapoint.output), 'reference': str(datapoint.reference)})
response = await ai.generate(model='googleai/gemini-flash-latest', messages=rendered.messages)
score = float(response.text.strip())
return EvalFnResponse(
test_case_id=datapoint.test_case_id or '',
evaluation=Score(score=score, status=EvalStatusEnum.PASS if score >= 0.5 else EvalStatusEnum.FAIL),
)
```
## Run evals via CLI
```bash
genkit eval:run datasets/my_dataset.json --evaluators=byo/my_eval
genkit eval:run datasets/my_dataset.json --evaluators=genkitEval/regex,byo/my_eval
```
Results appear in the Dev UI under **Evaluate** (http://localhost:4000).