fix: resolve CI linting and type errors
Some checks failed
CI / test (push) Has been cancelled

This commit is contained in:
2026-02-04 12:48:59 +00:00
parent 6f8f018f4f
commit d85bac7d65

View File

@@ -0,0 +1,97 @@
import asyncio
from typing import Any, Dict
import click
from promptforge.core.prompt import Prompt
from promptforge.core.template import TemplateEngine
from promptforge.core.config import get_config
from promptforge.providers import ProviderFactory
@click.command()
@click.argument("name")
@click.option("--provider", "-p", help="Override provider")
@click.option("--var", "-v", multiple=True, help="Variables in key=value format")
@click.option("--output", "-o", type=click.Choice(["text", "json", "yaml"]), default="text")
@click.option("--stream/--no-stream", default=True, help="Stream response")
@click.pass_obj
def run(ctx, name: str, provider: str, var: tuple, output: str, stream: bool):
"""Run a prompt with variables."""
prompts_dir = ctx["prompts_dir"]
prompts = Prompt.list(prompts_dir)
prompt = next((p for p in prompts if p.name == name), None)
if not prompt:
click.echo(f"Prompt '{name}' not found", err=True)
raise click.Abort()
variables = {}
for v in var:
if "=" in v:
k, val = v.split("=", 1)
variables[k.strip()] = val.strip()
template_engine = TemplateEngine()
try:
rendered = template_engine.render(
prompt.content,
variables,
prompt.variables,
)
except Exception as e:
click.echo(f"Template error: {e}", err=True)
raise click.Abort()
config = get_config()
selected_provider = provider or prompt.provider or config.defaults.provider
try:
provider_config: Dict[str, Any] = dict(config.providers.get(selected_provider, {}))
provider_instance = ProviderFactory.create(
selected_provider,
model=provider_config.get("model") if isinstance(provider_config, dict) else None,
temperature=provider_config.get("temperature", 0.7) if isinstance(provider_config, dict) else 0.7,
)
except Exception as e:
click.echo(f"Provider error: {e}", err=True)
raise click.Abort()
async def execute():
if stream:
full_response = []
async for chunk in provider_instance.stream_complete(rendered):
click.echo(chunk, nl=False)
full_response.append(chunk)
response = "".join(full_response)
else:
result = await provider_instance.complete(rendered)
response = result.content
click.echo(response)
if output == "json":
import json
click.echo("\n" + json.dumps({"response": response}, indent=2))
if prompt.validation_rules:
validate_response(prompt, response)
try:
asyncio.run(execute())
except Exception as e:
click.echo(f"Execution error: {e}", err=True)
raise click.Abort()
def validate_response(prompt: Prompt, response: str):
"""Validate response against rules."""
for rule in prompt.validation_rules:
if rule.type == "regex":
import re
if not re.search(rule.pattern or "", response):
click.echo("Warning: Response failed regex validation", err=True)
elif rule.type == "json":
try:
import json
json.loads(response)
except json.JSONDecodeError:
click.echo("Warning: Response is not valid JSON", err=True)