Skip to content

skill_agent

Level 2 · Tools and Skills — managing reusable skills with SkillStore.

A skill is a named, reusable instruction block the agent can pull into its prompt on demand. This example populates an InMemorySkillStore, filters skills by relevance, and formats a matched set for prompt injection — the three operations every skill-driven agent performs.

  • Lines: ~42
  • Prerequisites: none
  • Key concepts: SkillStore, Skill, filter_skills, format_skills_list

Run it

Installed via pip? Copy this example into your project with the CLI (no checkout needed):

pip install "symfonic-core[cli]"
symfonic examples add skill_agent
python -m skill_agent

Or, from a source checkout (run from the repo root):

python -m examples.skill_agent

Expected output:

Skills: ['summarize', 'translate', 'review']
Filtered 'code': ['review']
Prompt injection (<n> chars):
<formatted skill list>

Full code

"""Skill agent -- procedural knowledge chosen per turn, and read by the model.

Usage: python -m examples.skill_agent

Public surface used, and nothing else::

    from symfonic import Agent
    from symfonic.capabilities.prompting import PromptingCapability, StaticSource

The previous version of this file created three skills, filtered them,
formatted them, and printed the formatted string with the label "prompt
injection". Nothing injected it. No agent ran, so "the model uses the skill"
was a claim the example could not check -- and a skill that is selected,
formatted, and never reaches a prompt is indistinguishable from one that is.

Two things this rung makes visible instead:

* **Selection happens at composition, per turn.** ``SourceRequest`` carries
  the contribution, the scope and the turn -- not the user's question -- so
  the composition root is what narrows a corpus for a question, the same
  place ``knowledge_sources(query=...)`` takes one. Selection is a decision
  you can print and assert on before a model is ever called.
* **A skill is untrusted content.** It was written by an operator into a
  store, not authored inline by this deployment, so it arrives inside
  ``<untrusted-data>``. Text a store can be made to hold is text an attacker
  can propose; the wrapper is what keeps it from being an instruction.

The ``Skill`` shape below belongs to the example. The framework's seam is
``ContributionSource`` -- how a deployment stores and selects procedural
knowledge is its own decision, which is why nothing here inherits anything.
"""

from __future__ import annotations

import asyncio
from dataclasses import dataclass
from typing import Any

from symfonic import Agent
from symfonic.capabilities.prompting import PromptingCapability, StaticSource


@dataclass(frozen=True)
class Skill:
    """One procedure this deployment has learned."""

    name: str
    description: str
    content: str


SKILLS = (
    Skill("summarize", "Summarize text", "Extract the key points, drop the rest."),
    Skill("translate", "Translate text", "Convert between languages, keep register."),
    Skill("review", "Code review", "Find bugs before style; name the failing case."),
)


def select(skills: tuple[Skill, ...], question: str) -> tuple[Skill, ...]:
    """Which procedures this question needs.

    A substring match here; a retriever in your deployment. What matters is
    that the choice is made *before* the prompt is compiled and can be
    inspected -- a selection you cannot see is one you cannot debug the day
    it returns the wrong procedure.
    """
    words = set(question.lower().replace("?", "").split())
    return tuple(
        skill
        for skill in skills
        if words & set(f"{skill.name} {skill.description}".lower().split())
    )


def render(skills: tuple[Skill, ...]) -> str:
    return "\n".join(f"- {skill.name}: {skill.content}" for skill in skills)


def skilled_agent(provider: Any, skills: tuple[Skill, ...]) -> Agent:
    """One agent, carrying the procedures this turn selected.

    ``untrusted=True`` is the declaration that matters. A skill came from a
    store an operator writes to, so it is learned content and the compiler
    wraps it. Declaring it authored would let whatever is in that store speak
    with the deployment's own authority.
    """
    sources = []
    if skills:
        sources.append(StaticSource(text=render(skills), untrusted=True))
    return Agent(
        provider,
        instructions="Follow the procedures you were given.",
        capabilities=[PromptingCapability(sources=sources)],
    )


class RecordingProvider:
    """Keeps the prompts it was handed."""

    _symfonic_provider_family = "unknown"
    _symfonic_default_model = None

    def __init__(self, response: str) -> None:
        self.prompts: list[str] = []
        self._response = response

    def get_chat_model(self, config: Any) -> Any:  # noqa: ARG002
        from langchain_core.messages import AIMessage
        from langchain_core.outputs import ChatGeneration, ChatResult

        from symfonic.core.testing import MockChatModel

        prompts, response = self.prompts, self._response

        class _Model(MockChatModel):
            def _generate(self, messages, stop=None, run_manager=None, **kwargs):  # noqa: ANN001, ANN202
                prompts.append("\n".join(str(m.content) for m in messages))
                return ChatResult(
                    generations=[ChatGeneration(message=AIMessage(content=response))]
                )

            async def _agenerate(self, messages, stop=None, run_manager=None, **kw):  # noqa: ANN001, ANN202
                return self._generate(messages, stop, **kw)

        return _Model()

    def supports_thinking(self) -> bool:
        return False

    def supports_streaming(self) -> bool:
        return True

    def supports_forced_tool_choice(self, config: Any) -> bool:  # noqa: ARG002
        return True


def untrusted_block(prompt: str) -> str:
    opens, closes = "<untrusted-data", "</untrusted-data>"
    if opens not in prompt:
        return ""
    start = prompt.index(opens)
    return prompt[start : prompt.index(closes, start) + len(closes)]


async def main() -> None:
    print("== what this deployment knows how to do ==")
    for skill in SKILLS:
        print(f"  {skill.name:10s} {skill.description}")

    question = "Can you review this code?"
    chosen = select(SKILLS, question)

    print()
    print("== selected for this question ==")
    print(f"asked      : {question}")
    print(f"selected   : {', '.join(s.name for s in chosen) or '(none)'}")
    print("note       : decided before the prompt is compiled, so it can be")
    print("             printed and asserted on without calling a model.")

    provider = RecordingProvider("Looking for the failing case first.")
    result = await skilled_agent(provider, chosen).run(question)

    print()
    print("== what the model was actually given ==")
    for line in untrusted_block(provider.prompts[0]).splitlines():
        print(f"  | {line}")
    print(f"answered   : {result.text}")
    print("note       : untrusted, because a skill came from a store an")
    print("             operator writes to. Text a store can be made to hold")
    print("             is text an attacker can propose.")

    print()
    print("== a question no procedure covers ==")
    empty_question = "What is the freezer temperature?"
    none = select(SKILLS, empty_question)
    quiet = RecordingProvider("I have no procedure for that.")
    await skilled_agent(quiet, none).run(empty_question)
    print(f"selected   : {', '.join(s.name for s in none) or '(none)'}")
    print(f"in prompt  : {untrusted_block(quiet.prompts[0]) or '(no block)'}")
    print("note       : nothing selected, nothing rendered. An empty block")
    print("             would cost a budget slot and tell the model the store")
    print("             held nothing -- a different claim from not asking.")


if __name__ == "__main__":
    asyncio.run(main())

Step by step

1. Populate the skill store

store = InMemorySkillStore()
await store.create(Skill(name="summarize", description="Summarize text", content="Extract key points."))
await store.create(Skill(name="translate", description="Translate text", content="Convert languages."))
await store.create(Skill(name="review", description="Code review", content="Find bugs."))

A Skill has a name, a short description (used for matching), and a content block (the instruction injected into the prompt). InMemorySkillStore is the zero-infrastructure implementation of the SkillStore protocol — create is async because real stores hit a database.

2. Register the store as a dependency

deps = BaseAgentDeps(ModelProvider=MockModelProvider())
deps.register(SkillStore, store)

Just like documents and tools, the skill store is injected by protocol. An agent node asks deps for a SkillStore; it neither knows nor cares that it's the in-memory one.

3. List all skills

all_skills = await store.list()
print(f"Skills: {[s.name for s in all_skills]}")

list() returns every stored skill. This is what a "show my capabilities" query would enumerate.

4. Filter by relevance

matched = filter_skills(all_skills, "code")
print(f"Filtered 'code': {[s.name for s in matched]}")   # ['review']

filter_skills narrows a skill set to those relevant to a query string. Here only review (description "Code review") matches "code". In an agent turn you'd pass the user's message so the prompt carries just the pertinent skills — not the whole library — keeping token cost down.

5. Format for prompt injection

formatted = format_skills_list(all_skills)
print(f"Prompt injection ({len(formatted)} chars):\n{formatted}")

format_skills_list renders skills into the text block the prompt builder splices into the system prompt. This is the bridge between stored skills and what the model actually sees. In a full agent this is wired via a SkillsSection — see prompt_sections_lab.

The pattern

Store many → filter to the relevant few → format into the prompt. This keeps the prompt focused: the agent has access to a large skill library but only pays for the handful each turn actually needs.

What to try next

See also