Build an LLM-powered workflow with Akgents ~ akgentic-agent

Bringing the pieces together

#

I started this series with actors and typed messages. I then added durable teams, stepped aside to look at Pydantic AI, wrapped model calls in a ReactAgent, and turned functions and MCP servers into reusable tool cards. Each module made sense on its own, but the case-management sample still had two separate worlds: deterministic actors ran the workflow, while the LLM and tool examples lived beside it.

This episode joins those worlds with akgentic-agent. The module is the integration layer that combines an Akgent actor, the ReactAgent from akgentic-llm, and the ToolFactory from akgentic-tool in one BaseAgent.

That sounds like the moment to make every team member intelligent. I found the opposite lesson more useful: only the work that requires judgement should reach a model. Routing, storing facts, and enforcing approval remain deterministic. Judging the urgency of a case and drafting a plan become LLM work.

The result is a five-member team with two reasoning agents, two deterministic workflow agents, and one human bridge:

The complete code for this episode is in the basic-akgents sample. I link to the brnach I created, because both the framework and the sample continue to evolve.

Below is a youtube movie with a screencast that demos the application.

Watch the silent recording on YouTube

What akgentic-agent adds

#

The previous modules each owned one layer. akgentic-core supplied actors and messages, akgentic-llm supplied the ReAct loop and model lifecycle, and akgentic-tool supplied reusable capabilities. BaseAgent composes those layers into an LLM-powered actor.

flowchart TB
    MSG["Typed message"] --> BA["BaseAgent"]
    BA --> ACTOR["Akgent actor
mailbox, state, telemetry"]
    BA --> REACT["ReactAgent
model, context, limits"]
    BA --> FACTORY["ToolFactory
tools, context, commands"]
    REACT --> OUTPUT["Typed output"]
    OUTPUT --> HANDLER["Deterministic handler"]
    HANDLER --> NEXT["Next team message"]

The module’s standard collaboration path receives an AgentMessage, asks the model for a StructuredOutput, and lets the model choose recipients and message intentions. That is a useful fit for a conversational team in which the next step is open.

My case application is a workflow. Its routes are not open: a triage proposal goes to the coordinator, a human verdict goes back to triage, and an approved case goes to the executor. I keep those domain messages and ask the model only for the domain answer. BaseAgent.act() supports this without forcing the stock message and output types on the application.

Install the module directly when you want to manage its dependency versions yourself:

uv add akgentic-agent

Or use the framework meta-package for a set of module versions released and tested together:

uv add "akgentic-framework[agent]"

The sample used for this article locks akgentic-agent 1.7.0, akgentic-core 1.5.12, akgentic-llm 2.2.0, akgentic-team 1.5.6, and akgentic-tool 1.7.0. These packages are moving, so read the code together with its lock file.

Give judgement to a model, keep facts in code

#

The split between Akgent and BaseAgent is the most important concept in this episode.

@CaseCoordinator is a state machine. Its status tells it whether the next ResultMessage approves a priority or a plan. @CaseRepository owns a store and answers requests. Neither task improves when a model makes it probabilistic, slower, and more expensive.

Triage is different. The first version looked for words such as urgent. That was deterministic, but it was not judgement. An LLM can weigh the description, return a priority from a constrained enum, and explain its proposal. The executor has the same shape: it reads the case and returns a summary, ordered steps, and risks.

The team card shows the boundary. Only the two BaseAgent cards receive a prompt, model configuration, usage limits, and tools:

CASE_MODEL = ModelConfig(
    provider="openai",
    model="gpt-4o-mini",
    temperature=0.0,
)

CASE_RUN_LIMITS = RunUsageLimits(
    run_request_limit=10,
    tool_calls_limit=5,
)

CASE_AGENT_LIMITS = AgentUsageLimits(
    agent_request_limit=40,
    total_tokens_limit=200_000,
)

triage_agent_card = AgentCard(
    description="Judge how urgent a case is and propose a priority.",
    skills=["triage"],
    agent_class=CaseTriageAgent,
    config=CaseTriageConfig(
        name="@CaseTriage",
        role="Triage",
        prompt=PromptTemplate(template=TRIAGE_PROMPT),
        model_cfg=CASE_MODEL,
        run_usage_limits=CASE_RUN_LIMITS,
        agent_usage_limits=CASE_AGENT_LIMITS,
        tools=[CaseTool()],
    ),
)

This is also why there are two limit tiers. Run limits bound one ReAct loop. Agent limits accumulate over the lifetime of that actor. The default agent limits are unbounded, so I set both deliberately. The old usage_limits name is deprecated in akgentic-agent 1.7.0; new code should use run_usage_limits.

My rule of thumb is simple: a fact goes to a deterministic agent; a judgement goes to an LLM agent. The stored priority of a case is a fact. How urgent an untriaged description sounds is a judgement. That distinction saves tokens and makes the workflow easier to audit.

Subclass BaseAgent without losing its machinery

#

Turning the existing triage actor into a BaseAgent revealed a subtle inheritance trap. My first migration kept BaseConfig and BaseState. The class imported successfully, but it failed during on_start(): BaseAgent expects configuration fields such as prompt, model_cfg, tools, and usage limits.

Custom LLM-powered agents must extend AgentConfig and AgentState:

class CaseTriageConfig(AgentConfig):
    pass


class CaseTriageState(AgentState):
    known_case: bool = False
    proposed_priority: CasePriority = CasePriority.UNSET
    requester_id: str = ""
    approved: bool = False
    status: str = "new"

There is a second trap in on_start(). The base method creates the ReactAgent, builds the tool factory and command registry, and creates an AgentState. I then replace that state with the domain state, but I must preserve the framework fields:

class CaseTriageAgent(BaseAgent):
    config: CaseTriageConfig
    state: CaseTriageState

    def on_start(self) -> None:
        super().on_start()

        self.state = CaseTriageState(
            backstory=self.config.prompt.render(),
            tool_state=self.state.tool_state,
        )
        self.state.observer(self)

backstory is the rendered prompt. tool_state contains the tool layer’s persistent context-update baselines and block counter. Dropping it does not necessarily fail immediately; it quietly loses the state that lets restored agents continue their context updates correctly. I also avoid caching a reference to it, because restoring a team replaces the complete state object.

The same pattern powers @CaseExecutor. Once the pattern is correct, adding a second reasoning agent becomes pleasantly repetitive.

Keep the domain messages

#

The framework’s standard AgentMessage carries free-form content and an intention such as request, response, or notification. Its standard StructuredOutput lets a model decide which team member or role receives each next message.

I deliberately keep CaseTriageRequest, CaseTriageResponse, CasePriorityDecision, and the execution messages. They are the language of this workflow. The coordinator, not the model, decides where each one goes.

For a custom message to become input to act(), it only needs a rendering() method:

class CaseTriageRequest(Message):
    requester_id: str = ""
    case_id: str = ""

    def rendering(self) -> str:
        return (
            f"A case was handed to you for triage by "
            f"{self.requester_id or 'unknown'}.\n\n"
            f"The case id is `{self.case_id}`.\n\n"
            "Read the case with the `find_case` tool and report your assessment."
        )

That method satisfies the LlmRenderable contract. The message does not need a generic content field, and it does not need to inherit from AgentMessage.

I also leave out rendering_preview(). A message with both renderings can be shown to an agent while another run is in progress through the mailbox capability. A case-assignment message should wait for its own turn; I do not want a second case to bleed into the triage of the first one.

The output is equally specific:

class TriageOutput(BaseModel):
    known_case: bool = False
    already_prioritised: bool = False
    case_description: str = ""
    case_priority: CasePriority = CasePriority.UNSET
    reason: str = ""


def receiveMsg_CaseTriageRequest(
    self,
    message: CaseTriageRequest,
    sender: ActorAddress,
) -> None:
    output = self.act(message, TriageOutput)
    # Validate the domain result, update state, and send CaseTriageResponse.

Calling act(message, TriageOutput) still inherits the complete base policy: model configuration, tools, context, cancellation, capabilities, and usage limits. Only the input and output contracts are different.

One behaviour deserves an explicit guard. When a run is cancelled or a usage budget breaks, act() tries to return a default-constructed output. Both TriageOutput and ExecutionPlan therefore have defaults, and the handlers treat the empty instance as “no conclusion.” If the output model has a required field, constructing that fallback fails and the original error is raised. This is a design choice per agent, not a detail to discover during an incident.

A message is an assignment; a tool is freedom

#

Triage used to fetch a case by sending a request to @CaseRepository, waiting for the answer, and continuing from another handler. That is a good actor pattern for a deterministic workflow. Inside a ReAct loop it becomes unnecessary message ping-pong.

Both LLM agents now receive a read-only CaseTool. The model can call find_case, but the tool card does not expose its write functions unless allow_writes=True:

class CaseTool(ToolCard):
    backend: str = DEFAULT_CASE_REPOSITORY
    allow_writes: bool = False

    def get_tools(self) -> list[Callable[..., Any]]:
        tools = [self._find_case()]
        if self.allow_writes:
            tools.extend(self._write_tools())
        return tools

The distinction is more than implementation style. A CaseTriageRequest is an assignment the team has already decided to make. find_case is a capability the model may choose to use while completing that assignment.

The framework also gives a BaseAgent its intrinsic team and mailbox cards. The configured cards and the intrinsic cards feed several channels:

A callable on TOOL_CALL is not automatically a command. These are separate surfaces with separate permissions. That matters in the executor: the model can read the case, while the handler records an approved plan through an ordinary Python function. The model never receives a write tool.

Reuse the capabilities from the standalone ReAct agent

#

The earlier akgentic-llm sample added a priority capability and an observability capability directly to a ReactAgent. BaseAgent has an extra_capabilities() hook for exactly that work:

class CaseTriageAgent(BaseAgent):
    def extra_capabilities(self) -> list[AgentCapability[Any]]:
        return [CasePriorityCapability(), ObservabilityCapability()]


class CaseExecutorAgent(BaseAgent):
    def extra_capabilities(self) -> list[AgentCapability[Any]]:
        return [ObservabilityCapability()]

The executor does not judge priority, so it does not receive the priority capability. Capabilities remain role-specific even when the base class is shared.

There are two rules in this small override. I do not call super(), and I do not add the mailbox capability myself. The framework prepends that capability so cancellation stays first. The hook also runs while on_start() is building the internal ReactAgent, so it may read configuration but not objects that have not been constructed yet.

This reuse is one of my favourite outcomes of the refactor. The ReAct-level code from the previous episodes did not become throwaway learning material. It moved into real actors without changing its purpose.

Two model judgements, two human gates

#

The finished workflow has two halves. A new case passes through both. A case that already has a priority can enter at execution.

sequenceDiagram
    participant R as Runner
    participant C as @CaseCoordinator
    participant T as @CaseTriage
    participant H as @UserProxy / Human
    participant E as @CaseExecutor

    R->>C: HandleCaseRequest(stage="triage")
    C->>T: CaseTriageRequest
    T->>T: act(..., TriageOutput)
find_case tool
    T->>C: CaseTriageResponse
    C->>H: Approve the proposed priority?
    H->>C: ResultMessage
    C->>T: CasePriorityDecision
    T->>C: CaseTriageCompleted
    C->>E: CaseExecutionRequest
    E->>E: act(..., ExecutionPlan)
find_case tool
    E->>C: CaseExecutionProposal
    C->>H: Approve the plan?
    H->>C: ResultMessage
    C->>E: CaseExecutionDecision
    E->>C: CaseExecutionCompleted
    C-->>R: CaseClosed event

The handoff after priority approval is important. Earlier, the coordinator closed the workflow after triage. Adding @CaseExecutor to the team card was not enough; the coordinator needed a real seam between the two halves:

def receiveMsg_CaseTriageCompleted(
    self,
    message: CaseTriageCompleted,
    sender: ActorAddress,
) -> None:
    if message.approved:
        self._start_execution(
            message.case_description,
            message.case_priority,
        )
        return

    self._close(..., outcome="rejected", ...)

_start_execution() is shared by a newly triaged case and the execute <case id> entry point. On that direct route, the coordinator asks the deterministic repository actor for the stored description and priority. It does not spend two model calls asking triage to rediscover a fact.

It also refuses to execute an untriaged case. An agentic system is allowed to say no, and a rule in code is cheaper and more dependable than a sentence in a prompt asking the executor not to guess.

Put the approval boundary outside the model

#

The model proposes; the human decides; the handler writes. That sequence is the safety boundary of this sample.

Both CaseTool instances are read-only. After the human approves a plan, the executor handler calls record_execution() directly. A rejected plan follows the same path with approved=False and the human’s reason.

The full plan must survive between those two messages. The model run ends after CaseExecutionProposal; the human answers on a later turn. CaseExecutorState therefore retains the summary, steps, and risks, not just the line shown in the coordinator’s status:

self.update_state(
    {
        "proposed_plan": plan.summary,
        "proposed_steps": plan.steps,
        "proposed_risks": plan.risks,
    }
)

When the verdict arrives, the sample stores an ExecutionRecord and its audit line in one repository operation. Two writes could leave an executed case without the plan that was approved if the process failed between them.

I do not store a separate case-status field. Status is derived from priority and execution:

@property
def status(self) -> str:
    if self.execution is not None:
        return "executed" if self.execution.approved else "returned"

    return "new" if self.case_priority is CasePriority.UNSET else "triaged"

That removes a third fact that could drift away from the other two. It also gives the next executor useful memory. If a plan was returned, find_case renders the earlier plan and the human’s reason, so a later attempt can propose something different. The memory lives on the case, where it remains after the team has stopped.

One human bridge, two protocols

#

The coordinator is a plain Akgent, while triage and execution are BaseAgents. They speak to a person through different framework protocols.

• The coordinator sends a UserMessage and expects a ResultMessage.
• A BaseAgent sends an AgentMessage; HumanProxy answers it with another AgentMessage.

The CLI bridge therefore extends HumanProxy, but explicitly uses the original UserProxy method when it answers the coordinator:

def receiveMsg_UserMessage(
    self,
    message: UserMessage,
    sender: ActorAddress,
) -> None:
    self._io.say(sender.name, message.content)
    UserProxy.process_human_input(self, self._answer(), message)

Calling the overridden HumanProxy.process_human_input() on that channel would send the coordinator an AgentMessage. It has no handler for that message, so the workflow would simply wait.

The arrow that pointed the wrong way

#

The human bridge initially imported the terminal it printed to. A lint rule rejected that import: the console builds the team, while an agent inside the team reached back into the console. Moving the terminal module would have silenced the rule without fixing the dependency.

What the bridge actually needs is two verbs:

@runtime_checkable
class HumanIO(Protocol):
    def say(self, name: str, content: str, *, style: str = "msg.domain") -> None: ...

    def ask(self) -> str: ...

The bridge now depends on this protocol and loads the configured implementation from a dotted path:

DEFAULT_HUMAN_IO = "basic_akgents.cli.terminal.TerminalHumanIO"

The terminal still loads at runtime, but its name is configuration rather than an import edge. The value also survives serialization when a team is stopped and resumed. More importantly, the dependency points in the useful direction: a console or browser front end can implement HumanIO without the agents depending on either one.

Commands are messages that bypass the model

#

BaseAgent announces its command registry when it starts. In this team, the console listens for those CommandsAnnouncedEvents and can show commands such as /team_members, /stop, /compact, and /clear.

During a human question, a line such as this is forwarded to the named member:

@CaseTriage /clear

It arrives as an AgentMessage. Before calling the model, BaseAgent offers slash-prefixed content to its command registry. A recognised command runs in Python and its result becomes part of the agent’s context as a human action. The model sees on its next turn that the human cleared or compacted the conversation; its history did not mysteriously disappear.

An unknown slash command falls through to act() and costs a model call. That makes discovery important: the front end should show what the agents announced rather than maintain a second, hard-coded command list.

Observe the team from outside its lifetime

#

A live BaseAgent can report its own usage, but this console starts and stops many short-lived teams. The observer should outlive them.

UsageTracker subscribes to the same event stream used by the live feed. It retains each LlmUsageEvent and aggregates the events when the user types usage. AgentCommands does the same for command announcements. Neither service reaches into a live agent.

This completes a useful separation:

• domain messages drive the workflow;
• tool and model events make the reasoning visible;
• subscribers collect cross-team operational views;
• the event store preserves the team and its state;
• the case store preserves the approved business outcome.

Run the complete workflow

#

The sample needs Python 3.12 or 3.13, uv, and an OPENAI_API_KEY in the environment or .env file. Restore the locked environment and run the checks:

git clone https://github.com/jettro/basic-akgents.git
cd basic-akgents
git checkout blog_part_6

uv sync
make check
uv run pytest

Run a new case through triage and execution:

make run CASE=case_2

The application first asks whether priority 1, critical, should be approved. After approval, it hands the case to @CaseExecutor, which returns a structured plan with a summary, steps, and risks. The second question asks whether that plan may be executed.

I approved both questions in my run. The stored case then showed the result the human actually approved:

Case case_2
  priority  1 - critical
  status    executed
  plan      Escalate the down mail server to the IT support team
  decision  approved by jettrocoenradie

The exact prose changes between runs; the schema and approval boundary do not. The usage command reported two model responses for triage and two for execution: one response to call find_case, and one to turn the tool result into the structured domain output. Together they used 5,007 input tokens and 213 output tokens in this run, with an estimated cost of $0.0007.

Try the second entrance with a case that already has a priority:

make run CASE=case_4

# Or start the console and enter:
make run
execute case_4

The execute route performs the repository check without calling triage, then starts directly at the plan. Enter execute case_1 to see the deterministic refusal of an untriaged case. After a completed run, use case case_2, usage, commands, events, and team to inspect the business result and the framework telemetry from different angles.

Gotchas I would keep beside the code

#

This refactor exposed several details that are easy to miss:

• Extend AgentConfig and AgentState for a BaseAgent; BaseConfig and BaseState fail too late.
• Call super().on_start() before creating custom state, then preserve backstory and tool_state.
• Do not call super() from extra_capabilities() or add the mailbox capability again.
• Give custom messages a rendering() method; add rendering_preview() only if mid-run absorption is safe for that message.
• Decide whether your structured output can be default-constructed, because cancellation and limit handling depend on it.
• Remember that tool calls and commands are different channels, even if both start as Python callables.
• Load provider credentials before any BaseAgent starts. A deterministic team did not need them; this one does.
• A workflow can wait forever when no result is routed. Keep an explicit timeout and turn it into a visible outcome.

The larger lesson is that adding an LLM changes more than one handler. It changes configuration, state restoration, error paths, human messaging, usage accounting, and the meaning of persistence. akgentic-agent composes the necessary infrastructure, but the application still owns its domain boundaries.

What I take from this refactor

#

The case application now uses the complete stack from the series, but it has not become an LLM-controlled application. The model judges urgency and drafts a plan. Typed messages move those answers through a deterministic workflow. A person approves both decisions. Only then does ordinary code change the case.

That balance is what “agentic” means to me here. The reasoning is flexible where flexibility adds value; the routes, permissions, and business facts remain explicit. BaseAgent is useful because it brings the actor, ReAct, tool, context, command, and telemetry layers together without requiring the application to give the model control over all of them.

The result is not just a smarter team. It is a team whose judgement, control flow, human authority, cost, and durable outcome can each be inspected separately.