Pluggable task functions

Sentiment is one of five tasks the service needs to support. Entities, keywords, summary, and emotion round out the set. Rather than writing five unrelated functions, you write five functions with the same shape: async (text: str) -> dict | list | str. Same input, predictable outputs, easy to compose.

Five tasks, one signature

Each task is an async function that takes text and returns its own shape. The registry only knows the signature, not the body.

agent/tasks.py
python
async def extract_entities(text: str) -> list:
    prompt = f"""Extract named entities from the following text. Respond with ONLY
a JSON array (no prose, no markdown). Each item must have:
  - "text": the entity span verbatim from the text
  - "type": one of PERSON, ORG, LOCATION, PRODUCT, EVENT, DATE, OTHER

If there are no entities, return an empty array [].

Text:
\"\"\"{text}\"\"\"
"""
    raw = await _ask(prompt, max_tokens=400)
    data = _extract_json(raw)
    if not isinstance(data, list):
        return []
    out = []
    for item in data:
        if isinstance(item, dict) and "text" in item:
            out.append({"text": str(item["text"]), "type": str(item.get("type", "OTHER")).upper()})
    return out

Same recipe as classify_sentiment. Focused prompt, constrained output, defensive parsing, type normalisation. Writing the fifth task takes fifteen minutes because the template is already set.

agent/tasks.py
python
async def extract_keywords(text: str) -> list:
    prompt = f"""Extract the 5 to 10 most salient keywords or short key phrases
from the text. Respond with ONLY a JSON array of strings.

Text:
\"\"\"{text}\"\"\"
"""
    raw = await _ask(prompt, max_tokens=300)
    data = _extract_json(raw)
    return [str(k).strip() for k in data if str(k).strip()]


async def summarize(text: str) -> str:
    prompt = f"""Summarize the following text in 1-2 clear sentences. Respond
with ONLY the summary text itself. No preamble, no quotes, no markdown.

Text:
\"\"\"{text}\"\"\"
"""
    raw = await _ask(prompt, max_tokens=300)
    return raw.strip().strip('"').strip()


async def detect_emotion(text: str) -> dict:
    prompt = f"""Detect the dominant emotion. Respond with ONLY a JSON object of
shape {{"primary": joy|anger|sadness|fear|surprise|disgust|neutral,
"confidence": 0.0 to 1.0}}.

Text:
\"\"\"{text}\"\"\"
"""
    raw = await _ask(prompt, max_tokens=150)
    data = _extract_json(raw)
    return data

Three more tasks, same shape. Notice the max_tokens varies by task. Keywords need less room than entities. Summary needs more than emotion. Sizing tokens per task is a small but meaningful cost lever.