agentic-function

Turn LLM capabilities into ordinary Python functions with @agentic_function: describe the task in the docstring, constrain the output with a schema; the library handles prompt rendering, validation, retries, caching, tracing, and multi-backend execution.

Languages: English · 中文

• License: MIT
• Python: 3.10+
• Version: 0.0.1 PyPI

Installation

pip install agentic-function

Optional provider SDKs:

pip install "agentic-function[openai]"
pip install "agentic-function[anthropic]"
pip install "agentic-function[openai,anthropic]"

Editable install from source:

pip install -e ".[dev,openai,anthropic]"

Quick start

from agentic_function import agentic_function, AgenticResult, set_default_backend
from agentic_function.backends.mock_backend import MockBackend
from agentic_function.testing import mock_llm

set_default_backend(MockBackend())
mock_llm({"category": "positive", "confidence": 0.94, "reasoning": "..."})

@agentic_function(
    output_schema={
        "category": str,
        "confidence": float,
        "reasoning": str,
    },
)
def classify_sentiment(text: str) -> AgenticResult:
    """Classify the sentiment of ``text``.

    - ``category``: "positive" | "negative" | "neutral"
    - ``confidence``: [0.0, 1.0]
    - ``reasoning``: short explanation
    """

result = classify_sentiment("Amazing launch today!")
print(result.category, result.confidence, result.reasoning)
print(result.metrics.latency_ms, result.metrics.usage.prompt_tokens)

Runnable examples under examples/:

python examples/01_sentiment_classification.py
python examples/02_information_extraction.py
python examples/03_summarization.py
python examples/04_intent_routing.py
python examples/05_composition.py
python examples/06_real_minimax.py   # requires an API key

Examples 01–05 use MockBackend and need no API key.

Features

Usage

Schema and structured output

Declare output_schema as a dict, a BaseModel, or a Literal[...] return annotation. The library injects the JSON schema into the prompt (or uses provider tool / json_schema mode), applies common coercions, and validates with pydantic. On failure it retries according to policy. Callers receive typed fields, not a raw string to parse.

@agentic_function(output_schema={"label": str, "score": float})
def classify(text: str) -> AgenticResult:
    """Classify sentiment of ``text``."""

Composition

topic = extract_topic(article)
summary = make_summary(article, topic.topic, topic.tone)

See examples/05_composition.py.

Tool export

Export a function as OpenAI / Anthropic tool JSON for an external agent or custom tool loop:

from agentic_function import as_openai_tool, as_anthropic_tool, register, get_function

openai_tool = as_openai_tool(make_summary)
anthropic_tool = as_anthropic_tool(make_summary)

register(make_summary)
fn = get_function(make_summary.qualified_name)

Backends

Built-ins: MockBackend, OpenAIBackend, AnthropicBackend, and the MiniMax-CN preset minimax. Register custom backends with register_backend(...).

@agentic_function(backend="mock", output_schema={"label": str})
@agentic_function(backend="openai", model="gpt-4o-mini", output_schema={"label": str})
@agentic_function(backend="anthropic", model="claude-sonnet-4-20250514", output_schema={"label": str})
@agentic_function(backend="minimax", model="MiniMax-M3", output_schema={"label": str})

OpenAI-compatible servers (Ollama, vLLM, LocalAI, …)

Point OpenAIBackend at any Chat Completions–compatible endpoint via base_url. No separate Ollama adapter is required when the server speaks the OpenAI protocol:

from agentic_function import OpenAIBackend, register_backend, set_default_backend

ollama = OpenAIBackend(
    api_key="ollama",                      # required by the client; value often unused locally
    base_url="http://localhost:11434/v1",  # or OPENAI_BASE_URL
    default_model="llama3.2",
)
register_backend("ollama", ollama)
set_default_backend(ollama)

# Equivalent env-based setup with the built-in "openai" backend:
#   export OPENAI_API_KEY=ollama
#   export OPENAI_BASE_URL=http://localhost:11434/v1

@agentic_function(backend="ollama", model="llama3.2", output_schema={"label": str})
def classify(text: str):
    """Classify sentiment of ``text`` as label: positive|negative|neutral."""

Prompt parameters

Async

out = await classify.acall("terrible")

Works as free functions, methods, and classmethods (descriptor protocol).

Metrics, tracing, and cost

Every result includes CallMetrics:

result.metrics.latency_ms
result.metrics.usage.prompt_tokens
result.metrics.usage.completion_tokens
result.metrics.usage.total_tokens
result.metrics.cost_usd
result.metrics.cache_hit
result.metrics.attempts
result.metrics.retries
result.metrics.recovered
result.metrics.attempt_errors
result.metrics.timings
result.metrics.total_cost_usd

Tracing, budgets, and aggregation:

from agentic_function import (
    trace,
    Budget, BudgetTracker, install_budget_tracker,
    Aggregator, install_default_aggregator,
)

with trace("nightly_eval") as ctx:
    out = classify(sample.text)
    ctx.span.set_attribute("sample.id", sample.id)

install_budget_tracker(BudgetTracker(budgets=[
    Budget(metric="cost_usd", limit=5.0),
]))

agg = install_default_aggregator(Aggregator())
print(agg.summary())
print(agg.to_prometheus())

Diagnostics:

from agentic_function import diagnose, explain_failure, snapshot

print(diagnose(result).to_dict())
print(explain_failure(exc))
print(snapshot(result))

Cache keys cover (model, schema, few_shots, prompt_hash). Bound retries with RetryPolicy(max_retries=..., base_delay=..., max_delay=...).

examples/06_real_minimax.py runs against a live backend (API key required).

Errors and retries

A successful call returns a value that matches output_schema. Transient failures are retried according to RetryPolicy; if every attempt fails, the call raises.

call → prompt → cache lookup
                   │ miss
                   ▼
              backend request
                   │
          ┌────────┴────────┐
          │ OK              │ BackendError / ParseError / ValidationError
          ▼                 ▼
     coerce + validate   backoff retry
          │                 │
          │ valid           │ retries exhausted
          ▼                 ▼
       return            RetryExhaustedError

BackendError.status_code carries the HTTP status when the provider SDK exposes it (for example 429). error_category_of(exc) returns "rate_limit" for status 429, otherwise "backend" for other BackendErrors.

Library errors subclass AgenticFunctionError. On exhaustion, RetryExhaustedError exposes last_exception, metrics, and attempt_errors:

from agentic_function import BackendError, RetryExhaustedError, error_category_of

try:
    result = classify(text)
except RetryExhaustedError as exc:
    cause = exc.last_exception
    status = getattr(cause, "status_code", None)
    if error_category_of(exc) == "rate_limit" or status == 429:
        # apply caller-side backoff / queueing
        ...
    raise

Control retries with max_retries or retry_policy=RetryPolicy(...). Application code may catch failures and apply its own fallback; the library does not substitute default schema fields on failure.

Credentials

Set provider keys via environment variables (for example OPENAI_API_KEY, ANTHROPIC_API_KEY, MINIMAX_CN_API_KEY) or pass them when constructing a backend. Do not place secrets in source files or pass them through configure().

Testing

from agentic_function.testing import mock_llm, mock_llm_table, freeze_time, capture_metrics

mock_llm({"label": "positive", "score": 0.95})
assert classify("amazing").label == "positive"

mock_llm_table([
    {"label": "positive", "score": 0.9},
    {"label": "negative", "score": 0.8},
])

with freeze_time():
    classify("text")

with capture_metrics() as bag:
    classify("a")
    classify("b")
assert len(bag) == 2

You can also use MockBackend, or patch the openai / anthropic clients to assert request shaping (see tests/test_anthropic_backend.py).

pytest tests/ -m "not live_llm"          # offline
pytest tests/ -m live_llm -v             # live MiniMax (API key required)
pytest --cov=agentic_function

Live tests honor AGENTIC_LIVE_LLM_PACE and AGENTIC_LIVE_CALL_GAP (seconds) to space requests.

Isolation points:

Decorator parameters

Environment variables: AGENTIC_FUNCTION_MODEL, AGENTIC_FUNCTION_BACKEND, AGENTIC_FUNCTION_CACHE, AGENTIC_FUNCTION_CACHE_DIR, plus OPENAI_API_KEY, ANTHROPIC_API_KEY, MINIMAX_CN_API_KEY, etc.

Public API

from agentic_function import (
    agentic_function, AgenticFunction, AgenticResult, DynamicResult,
    SchemaSpec, resolve_schema, render_prompt,
    LLMBackend, LLMResponse, StreamChunk,
    MockBackend, OpenAIBackend,
    register_backend, get_backend, get_default_backend, set_default_backend,
    known_backends,
    Executor, GlobalConfig, configure, global_config,
    TraceContext, TraceSpan, TraceRecorder, trace, get_current_trace,
    CallMetrics, TokenUsage, PhaseTimings,
    RetryPolicy, default_retry_policy,
    CacheBackend, InMemoryCache, DiskCache, NullCache,
    get_default_executor, set_default_executor,
    Budget, BudgetTracker, BudgetExceededError,
    install_budget_tracker, get_default_budget_tracker,
    Aggregator, FunctionStats,
    install_default_aggregator, get_default_aggregator,
    Diagnostic, diagnose, diagnose_metrics, explain_failure, snapshot,
    FunctionRegistry, get_global_registry, register, get_function,
    as_openai_tool, as_anthropic_tool,
    testing,
    AgenticFunctionError, BackendError, CacheError, CompositionError,
    ConfigError, ParseError, RegistrationError, RetryExhaustedError,
    SchemaError, TimeoutError_ as TimeoutError, ValidationError,
    error_category_of,
)

AnthropicBackend and MiniMax live under agentic_function.backends and register as "anthropic" / "minimax".

Architecture

@agentic_function(...)
        │
        ▼
  AgenticFunction  (descriptor / call / await)
        │ ExecutionRequest
        ▼
     Executor
   trace → cache → retry → schema → backend
        │
        ├── BudgetTracker
        ├── Aggregator
        └── as_*_tool / Registry

Project layout

agentic-function/
├── agentic_function/
│   ├── core/
│   ├── backends/
│   ├── runtime/
│   ├── composition/
│   ├── validation/
│   ├── utils/
│   └── testing.py
├── tests/
└── examples/

Roadmap

Current release: 0.0.1.

In tree today

• Decorator API + pydantic / dict / Literal schemas
• Pluggable backends (Mock, OpenAI, Anthropic, MiniMax)
• OpenAI-compatible endpoints via OpenAIBackend(base_url=...) (Ollama, vLLM, …)
• Composition + tool export (as_openai_tool / as_anthropic_tool)
• Cache, cost estimates, mock_llm test helpers
• Async, tracing, budget / aggregator / diagnostics
• HTTP status_code on BackendError (incl. rate-limit categorization)

Next

• Streaming output as a stable public API
• Broader CI and typed packaging polish
• 1.0 — longer-term API stability guarantee

Contributing

See CONTRIBUTING.md.

License

MIT