Alms Langgraph Agent

ALMS LangGraph Agent

Use this skill to build agentic FastAPI services in the optimized ALMS style: API endpoints stay thin, usecases orchestrate business flow and job lifecycle, actions execute compiled workflows, and the agent layer owns prompts, structured outputs, tools, state, nodes, and LangGraph wiring.

Treat alms as the canonical architecture. Treat production ALMS implementations as evidence for reusable patterns: long-running jobs, item-level ledgers, deterministic fast paths, approved memory, retrieval-backed reasoning, conflict checks, human review, rule hardening, dashboard/status APIs, and clean failure handling.

Do not copy a source repo''s domain, filenames, docs paths, examples, endpoint names, or business nouns unless the target repo already uses them. Extract the mechanism, then adapt it to the target business problem.

For detailed conventions and templates, read references/alms-patterns.md when adding or changing code.

Profile / Capability Contract

Always read `[tool.alms]` from `pyproject.toml` before adding or changing code.

ALMS projects declare their active profile and capabilities in pyproject.toml:

[tool.alms]
profile = "core-api"
capabilities = ["runtime_auth", "tests"]

The skill must honour these constraints. If [tool.alms] is missing, infer conservatively by inspecting what already exists:

• If src/agents/workflows/ has feature folders, treat as workflow-agent or full.
• If src/providers/ai/ exists, treat as at least llm-agent.
• If src/database/ exists, treat as at least db-agent.
• Otherwise treat as core-api.
• Never assume `full` unless clearly indicated.

Profile rules

Constraint enforcement

• Do not add imports that require optional extras unless the capability is present.
• Do not create folders under src/agents/, src/providers/ai/, src/database/, or src/observability/ unless the matching capability is enabled.
• Do not register routers for disabled capabilities (e.g. no /metrics route without observability).
• Do not make tests import optional systems in core-api profile.
• Do not add dependencies to `pyproject.toml` outside the active capability set.

Do Not Overbuild

Use the minimum architecture the task actually requires — within the project''s capability boundaries.

For a simple task in an llm-agent profile, a thin action calling a structured agent is enough:

Endpoint -> UseCase -> Action -> Structured Agent

For production tasks in a workflow-agent or full profile, use the full workflow:

Endpoint -> UseCase -> Action -> LangGraph Workflow -> Nodes / Tools / Agents

Only add the production workflow when the task involves AND the langgraph capability is enabled:

• batch processing or long-running jobs
• auditability or traceable decisions
• source anchoring the model must not override
• expensive recomputation worth caching in approved memory
• human review
• conflict detection
• approved memory lookup
• deterministic rule promotion
• status or display APIs
• coverage validation that catches missing or duplicate output

Do not add ledger, approved memory, conflict checks, or human review unless the task type in this table says to. Do not add LangGraph at all unless the langgraph capability is enabled.

Core Workflow

1. Read the project profile before writing any code:

• Inspect pyproject.toml for [tool.alms].
• Note the profile and capabilities list.
• If [tool.alms] is absent, infer from existing files:
• src/agents/workflows/ -> workflow-agent or full
• src/providers/ai/ -> llm-agent
• src/database/ -> db-agent
• else -> core-api
• Then inspect the existing repo shape:
• architecture, project-structure, guideline, README, rules, or planning docs if they exist
• pyproject.toml, lockfiles, test config, and app entrypoints
• src/api/endpoints/v1/
• src/execution/usecases/
• src/execution/actions/
• src/config/
• existing tests for the same feature or layer
• Only inspect folders that match enabled capabilities:
• src/agents/agent_manager/ — only if llm is enabled
• src/agents/prompts/ — only if llm is enabled
• src/agents/schemas/ — only if llm is enabled
• src/agents/tools/ — only if llm or langgraph is enabled
• src/agents/workflows/ — only if langgraph is enabled
• src/providers/ai/ — only if llm is enabled
• src/database/ — only if database is enabled
• src/observability/ — only if observability is enabled

2. Choose the right workflow depth within capability limits:

• For core-api: use plain FastAPI + usecase/action. No agents, no AI.
• For llm-agent without langgraph: use a thin action around a structured-output agent.
• For workflow-agent or full: use stateful workflows with ledger, deterministic fast paths, retrieval or LLM reasoning, coverage validation, conflict checks, summary/reconciliation, and human review.
• For long-running work in any AI-capable profile: expose process, status, and display-friendly result endpoints, and use background tasks or a queue instead of blocking the request.

3. Add features layer by layer only within enabled capabilities:

• Endpoint: validate HTTP input, inject the usecase, enqueue background work when needed, return AppResponse or the repo response wrapper.
• Usecase: own job creation, status transitions, orchestration, retries, batching, dashboard/display payloads, and clean failure recording.
• Action: lazily build and cache the compiled workflow (langgraph only), adapt inputs, call ainvoke, and normalize the workflow result.
• Workflow: (langgraph only) build StateGraph, add nodes, wire edges, compile only. Keep feature-sized workflows in src/agents/workflows/<feature>/.
• State: (langgraph only) define durable state in state.py; include input, ledger, reports, outputs, final status, and errors.
• Nodes: (langgraph only) run deterministic tools first, call structured agents only when needed, validate coverage, return partial state, and preserve audit evidence.
• Agents and schemas: (llm only) keep Pydantic structured outputs in feature schema modules such as src/agents/schemas/<feature>.py; register them in AgentManager.
• Prompts: (llm only) store system prompts as markdown files under src/agents/prompts/agents/ and lazy-load them through PromptManager.
• Tools: (llm or langgraph) keep retrieval, memory lookup, review persistence, and deterministic rule runners behind tool classes.

4. Preserve ALMS dependency flow:

• API can depend on Execution.
• Execution can depend on Agents, Providers, Repositories, and Utils.
• Agents can depend on Providers, Config, and Utils.
• Tools may use repositories or sessions when they are persistence adapters.
• Do not make endpoints call LangGraph, model loaders, prompt files, tools, or repositories directly.

Production Reliability Ladder

(Requires langgraph capability.)

Use this production ALMS pattern when the output must be auditable, reviewable, or expensive to recompute:

preprocess ledger
-> deterministic safe rule check
-> exact approved memory lookup
-> retrieval-backed LLM reasoning
-> row/item coverage retry or held state
-> deterministic conflict checks
-> optional LLM conflict evidence
-> summary and unit/result reconciliation
-> human review queue
-> ledger/final output write
-> approved memory after accept/override
-> safe DSL rule candidate after repeated proof

Important guardrails:

• No silent loss: every input row/item must be represented in ledger, coverage report, final output, held state, or review evidence.
• Source anchoring: preserve source units, source category, source identifiers, and other user/provider facts. LLMs may classify; they should not silently rewrite the source of truth.
• Exact memory first: approved memory should use stable signatures and hashes. Avoid fuzzy matching in production logic unless the user explicitly accepts that risk.
• Human review is the trust boundary: LLM output can propose decisions; accepted or overridden human decisions create approved memory.
• Deterministic rules come late: promote repeated approved memory into safe DSL or inspectable rules only after enough clean evidence.
• Safe rule execution: prefer JSON/DSL conditions over arbitrary generated Python.
• Status is sacred: job status must reflect workflow result (completed, human_review, failed, etc.), not a hardcoded happy path.
• Clean failure handling: after a background job is marked failed and committed, do not re-raise into the ASGI stack.
• PII minimization: review evidence should store allowlisted context by default, with raw rows behind explicit settings.

Naming Style

Prefer the repo''s explicit business names over generic abstractions:

• Endpoint file: <feature>.py or process_<thing>.py
• Endpoint route group: /api/v1/<feature>/...
• Usecase file/class: process_<thing>_usecase.py, Process<Thing>UseCase
• Action file/class: process_<thing>_action.py, Process<Thing>Action
• Workflow package: src/agents/workflows/<thing>/
• Workflow builder: build_<thing>_workflow
• Workflow state: <Thing>WorkflowState or <Thing>State
• Node function: <step_name> for workflow steps or llm_call_<agent_or_step> for direct LLM nodes
• Result schema: <Thing>Result, <Thing>ChunkResult, or domain-specific Pydantic models
• Prompt file: agent_<thing>.md
• PromptManager property: <thing>
• AgentManager property: <thing>
• Tool class: <Thing>Tool, <Thing>MemoryTool, <Thing>QueueTool, <Thing>RuleRunner

LangGraph Defaults

(Requires langgraph capability.)

Use StateGraph when the workflow has explicit state, routing, retries, fan-out, deterministic checks, human review, or multiple LLM steps. Use simple structured-output calls only for small single-agent routes.

Keep state schemas readable with TypedDict for graph state and Pydantic BaseModel for structured LLM output. When parallel workers aggregate results, use Annotated[list, operator.add].

Use feature folders for real workflows:

src/agents/workflows/<feature>/
  state.py
  nodes.py
  build.py

Use Send("node_name", custom_state) for map-style fan-out. Use conditional edges when routing chooses the next node. Use Command only when a node must both update state and route.

Compile workflows in build.py, but invoke them from action classes. Cache the compiled graph on the action instance with self._workflow or a workflow property.

Keep graph-level retry small for LLM calls. Put business retries in usecases or node helpers where failed outputs can be normalized into held/review state.

LangChain Defaults

(Requires llm capability.)

Use a LangchainModelLoader or get_llm() provider helper to centralize model configuration from settings. Keep provider details out of node functions.

Use AgentManager to lazy-load and cache structured-output agents:

self.model.with_structured_output(OutputSchema)

For tools plus structured output, use a tool-aware agent only when the tool has deterministic value. Otherwise, retrieve tool context in the node, pass it into the prompt, and keep the LLM output schema simple.

Factory Compatibility

ALMS may use function-based agent factories alongside or instead of AgentManager.

@lru_cache(maxsize=1)
def create_sample_agent() -> Any:
    """Build the sample agent lazily so non-agent routes do not depend on AI setup."""
    ...

Do not blindly replace existing factories.

• If the repo uses create_*_agent() factories, keep them for backward compatibility.
• Use AgentManager for new production structured-output workflows.
• Add AgentManager alongside existing factories if both styles are needed.
• Refactor old factories to AgentManager only when explicitly requested.

Future Domain-Module Compatibility

ALMS currently uses a layer-first directory structure:

src/api/endpoints/v1/<feature>.py
src/execution/usecases/<feature>_usecase.py
src/execution/actions/<feature>_action.py
src/agents/workflows/<feature>/

A future ALMS version may support a domain-module structure:

src/modules/<feature>/
  api.py
  schemas.py
  usecases.py
  actions.py
  workflows/
    state.py
    nodes.py
    build.py

The dependency flow remains the same regardless of directory structure:

API -> UseCase -> Action -> Workflow / Agent / Provider

When working in a repo that has not migrated, keep the layer-first structure. Do not reorganize into domain modules unless explicitly requested.

Run And Verify

Prefer the repo''s existing commands. Common ALMS commands are:

uv sync
uv run uvicorn src.api.main:app --port 3000 --reload
uv run pytest src/tests
uv run pytest src/tests/v1 -v
uv run ruff check src
uv run ruff format src

Profile-specific verification

For core-api — the app must start without any optional dependencies:

uv sync
uv run pytest src/tests
python -c "import src.api.main; print('core import ok')"

Core profile must NOT require: langchain, langgraph, sqlalchemy, asyncpg, redis, opentelemetry, prometheus-client, scalar-fastapi.

For llm-agent:

uv sync --extra ai --extra docs
uv run pytest src/tests

For workflow-agent:

uv sync --extra ai --extra workflow --extra docs
uv run pytest src/tests

For db-agent:

uv sync --extra db
uv run pytest src/tests

For observable:

uv sync --extra observability
uv run pytest src/tests

For full (current v0.3 behaviour):

uv sync --extra full
uv run pytest src/tests

Production agent workflow verification

(Requires langgraph capability.)

For production agent workflows, also verify:

• Workflow compiles.
• Action can invoke the workflow with a minimal payload.
• Coverage validation catches missing, duplicate, and extra outputs.
• Deterministic fast paths still pass coverage and conflict checks.
• Human review paths persist enough evidence.
• A second run can hit approved memory or code rules only when exact coverage exists.
• Status/display APIs show completed, human_review, and failed clearly.

Final Response Format

After any changes, report in this format:

## Summary
What changed and why.

## Files Changed
List each file and the purpose of the change.

## Architecture Impact
How the change affects ALMS boundaries:
API, UseCase, Action, Agent/Workflow, Provider, Database, Observability, Config.

## Profile Impact
Which capabilities were used, and whether any capability boundaries were crossed or needed.

## Verification
Commands run, for example:
- uv sync
- uv run pytest src/tests
- uv run ruff check src
- uv run ruff format src

If a command was not run, say so clearly.

## Known Limitations
Incomplete work, assumptions, or follow-up risks.

## Next Recommended Step
One clear next step.

Changelog

Unreleased (0.4.0)

• Skill is now profile-aware: reads [tool.alms] from pyproject.toml before generating code.
• Added "Profile / Capability Contract" section in SKILL.md with capability rules.
• Updated "Core Workflow" to start with profile inspection and capability-aware directory scanning.
• Updated "Do Not Overbuild" to include capability-based constraints.
• Made "Project Shape" section in references/alms-patterns.md conditional on active capabilities.
• Added capability gates to all production workflow recipes in references/alms-patterns.md.
• Updated "Run And Verify" with profile-specific verification commands.
• Added "Profile Impact" field to Final Response Format.
• Bumped version to 0.4.0. Compatible with alms >=0.3.0.
• Backward compatible: full profile preserves all v0.3.0 behaviour.

0.3.0 - 2026-06-16

• Added Do Not Overbuild section with Simple Agent vs Production Workflow decision table.
• Added Factory Compatibility section documenting create_*_agent() coexistence with AgentManager.
• Added Future Domain-Module Compatibility section for layer-first to domain-module migration path.
• Added Final Response Format section for AI coding agent output.
• Added Do Not Overbuild decision table and factory compat note to references/alms-patterns.md.
• Bumped skill version to 0.3.0. Compatible with alms >=0.2.1.

0.2.1 - 2026-05-12

• Shortened the skill frontmatter description for more precise activation.
• Added version and compatible_with metadata.
• Documented compatibility with alms >=0.2.1.
• Clarified that missing prompt manager, markdown prompt, schema, or workflow skeleton directories should be created before implementing production LangGraph behavior.

alms-langgraph-agent

ALMS-style LangGraph and LangChain agent builder skill for AI coding agents.

This skill captures KJ''s optimized ALMS architecture for agentic FastAPI services. It uses alms as the canonical project structure and extracts reusable production patterns from real ALMS implementations without hardcoding one source repo or domain.

Install

npx skills add KJ-AIML/alms-langgraph-agent-skill

Compatibility

• Skill version: 0.4.0
• Compatible with: alms >=0.3.0

Use this public skill for LangGraph/LangChain agent workflows. Use the bundled .agents/skills/alms-dev skill from the alms repo for normal backend changes such as endpoints, usecases, actions, repositories, providers, settings, middleware, tests, and docs.

Profile Support (new in 0.4.0)

This skill is profile-aware. It reads [tool.alms] from the target project''s pyproject.toml and constrains code generation to the active capabilities.

Supported profiles:

The skill will never add imports, folders, routers, or tests for capabilities that are not enabled.

What It Does

• Builds LangGraph/LangChain agent services in the ALMS style
• Keeps FastAPI endpoints thin and pushes orchestration into usecases
• Uses actions to lazily build and invoke compiled LangGraph workflows
• Places agent code under src/agents/ with managers, prompts, schemas, tools, and feature-scoped workflows
• Uses markdown prompt files through a PromptManager
• Uses structured Pydantic outputs through an AgentManager
• Preserves KJ''s naming style for process_*, build_*_workflow, workflow nodes, and action/usecase layers
• Adds production guardrails: ledgers, exact approved memory, deterministic safe rules, retrieval-backed reasoning, coverage retry, conflict checks, human review, summary reconciliation, and status/display APIs

Architecture Preference

Treat alms as the optimized source of truth. For simple tasks, a short chain is enough:

API Endpoint -> UseCase -> Action -> Structured Agent

For production workflows that require state, auditability, long-running jobs, or multi-step reasoning:

API Endpoint -> UseCase -> Action -> LangGraph Workflow -> Agent Manager / Prompt Manager / Tools

See SKILL.md "Do Not Overbuild" for the full decision table on when to use each path.

Extract these production implementation patterns when the target problem needs them:

• background mapping jobs
• feature-scoped workflow package: state.py, nodes.py, build.py
• AgentManager and PromptManager
• structured output schemas
• PageIndex/retrieval-backed reasoning
• approved-memory lookup before LLM reasoning
• safe DSL rule runner before memory and LLM reasoning
• row coverage validation and held state
• deterministic plus optional LLM conflict checks
• summary aggregation and result reconciliation
• human review queue and review decision APIs
• dashboard/status/display/SSE endpoints

When alms and a production repo disagree, prefer the target repo''s current shape for low-churn maintenance, and prefer alms for new projects.

If the target ALMS repo does not yet contain src/agents/prompts/prompt_manager.py, src/agents/prompts/agents/, src/agents/schemas/, or feature-scoped workflow folders, create that skeleton before adding production agent behavior only if the `llm` capability is enabled.

Project Structure

The skill expects and reinforces the ALMS layer structure. The exact files depend on the active profile (see SKILL.md "Profile / Capability Contract").

src/
  api/
    endpoints/
      v1/
        <feature>.py
        schemas/
  execution/
    actions/
      process_<thing>_action.py
    usecases/
      process_<thing>_usecase.py
  agents/              # only when llm is enabled
    agent_manager/
      agent.py
    prompts/
      prompt_manager.py
      agents/
        agent_<thing>.md
    schemas/
      <feature>.py
    tools/              # only when langgraph is enabled
      <feature>_tool.py
    workflows/          # only when langgraph is enabled
      <feature>/
        state.py
        nodes.py
        build.py
  providers/
    ai/                 # only when llm is enabled
      base.py
      factory.py
      langchain_model_loader.py
  database/             # only when database is enabled
  observability/        # only when observability is enabled
  config/

Usage Examples

User: "Add an auditable document classification workflow in this ALMS repo"
-> Checks [tool.alms] for langgraph capability, creates schemas, prompts, AgentManager registry, tools, workflow state/nodes/build, action, usecase, endpoints, status/display APIs, and tests.

User: "Refactor this LangGraph agent to follow my ALMS style"
-> Moves orchestration into usecases/actions, centralizes prompts and model loading, and keeps endpoints thin.

User: "Add a simple health endpoint in this core-api project"
-> Creates a plain FastAPI endpoint with usecase/action. No agents or AI imports.

Files

• SKILL.md - main skill instructions, profile contract, and trigger metadata
• references/alms-patterns.md - detailed implementation patterns with capability gates
• agents/openai.yaml - optional UI metadata for compatible agents
• CHANGELOG.md - compatibility and release notes

License

MIT

ALMS LangGraph Agent Patterns

These conventions are anchored in alms, which is the optimized target architecture. They also extract reusable lessons from production ALMS implementations: durable jobs, audit trails, deterministic fast paths, approved memory, retrieval-backed reasoning, human review, and rule hardening.

When the target repo already has a local pattern, follow it unless changing it clearly improves correctness. Use alms for the base structure and use production evidence for guardrails, not for copying repo-specific names or domain concepts.

Do not turn one source repo''s domain concepts into generic rules. Use real implementations as examples of the deeper mechanism, then adapt names, paths, prompts, schemas, endpoints, and tests to the target repo.

Always read `[tool.alms]` from `pyproject.toml` before applying any pattern. The active capabilities determine which sections of this reference apply. See SKILL.md "Profile / Capability Contract" for the full rules.

Do Not Overbuild

Before adding workflow state, ledger, memory, or review, check which row the task falls under AND whether the required capabilities are enabled:

The production feature recipe below applies to the full production workflow. Scale back for simpler tasks by removing what the table row does not require. Skip the recipe entirely if `langgraph` is not enabled.

Mental Model

Use this direction (adapted to active capabilities):

API endpoint
-> Execution usecase
-> Execution action
-> (if langgraph) LangGraph workflow -> Nodes -> AgentManager / PromptManager / Tools / Providers
-> (if llm only) Structured Agent -> AgentManager / PromptManager / Provider

For simple work, the chain can be short. For production decisions with langgraph, add a reliability ladder around the model:

preprocess ledger
-> code rule check
-> approved memory check
-> retrieval-backed LLM reasoning
-> coverage retry / held state
-> conflict check
-> summary and reconciliation
-> human review queue
-> final output write

The central lesson is that the LLM is one decision step inside an auditable system, not the system itself.

Project Shape

The project structure depends on which capabilities are active.

Always present (core-api and all profiles)

src/
  api/
    main.py
    endpoints/
      v1/
        dependencies.py
        routers.py
        health.py
        schemas/
        <feature>.py
    middlewares/
      error_handler.py
      logging.py
      security.py
    router/
      routers.py
  execution/
    actions/
      <feature>_action.py
    usecases/
      <feature>_usecase.py
  config/
    settings.py
    logs_config.py
  core/
    exceptions.py
  tests/
    conftest.py
    v1/
      test_health.py

When llm capability is enabled

src/
  agents/
    agent_manager/
      agent.py or agent_manager.py
    prompts/
      prompt_manager.py
      agents/
        agent_<thing>.md
    schemas/
      <feature>.py
  providers/
    ai/
      base.py
      factory.py
      langchain_model_loader.py

When langgraph capability is enabled (adds to llm)

src/
  agents/
    tools/
      <feature>_tool.py
    workflows/
      <feature>/
        state.py
        nodes.py
        build.py

When database capability is enabled

src/
  database/
    connection.py
    models/
    repositories/
      base.py

When observability capability is enabled

src/
  observability/
    __init__.py
    metrics.py
    tracing.py
  api/
    endpoints/
      v1/
        metrics.py
    middlewares/
      observability.py

When redis capability is enabled

src/
  providers/
    cache/

For small starters, a flat src/agents/workflows/build.py and nodes.py is acceptable. For real production workflows, use a feature folder like src/agents/workflows/mapping/ so state, nodes, and graph wiring stay navigable.

Prefer the ALMS provider boundary for new or refactored projects: use get_ai_provider() from src/providers/ai/factory.py. This returns a configured AIModelProvider. Inject models through provider.get_chat_model(tier) rather than instantiating LangchainModelLoader directly. This keeps nodes and agents vendor-neutral — provider selection stays in settings, not in agent code.

If a starter repo is older than alms 0.3.0 and lacks src/agents/prompts/prompt_manager.py, src/agents/prompts/agents/, src/agents/schemas/, or feature-scoped workflow folders, add those skeleton directories before implementing production behavior only if the `llm` capability is enabled.

Production Feature Recipe

Only use this recipe when `langgraph` is enabled. For llm-agent without langgraph, skip steps 7, 8 (workflow state/nodes/build). For core-api, skip steps 5-11 entirely (use plain usecase/action).

To add a production agent workflow called <thing>:

1. Inspect existing docs, rules, [tool.alms], entrypoints, and layer patterns; do not assume a fixed documentation filename exists. 2. Define endpoint request/response schemas in src/api/endpoints/v1/schemas/<thing>.py or locally for tiny endpoints. 3. Add an endpoint in src/api/endpoints/v1/<thing>.py. 4. Add or reuse dependencies in src/api/endpoints/v1/dependencies.py. Guard AI/database imports with try/except ImportError if the capability is optional. 5. Add a usecase that owns job lifecycle, orchestration, status/display payloads, and failure recording. 6. Add an action that lazily builds and invokes the compiled workflow (or calls the agent directly if no langgraph). 7. (langgraph only) Add a workflow package: src/agents/workflows/<thing>/state.py, nodes.py, build.py. 8. (llm only) Add Pydantic structured output schemas under src/agents/schemas/<thing>.py. 9. (llm only) Add prompt markdown files under src/agents/prompts/agents/. 10. (llm only) Register agents in AgentManager and prompts in PromptManager. 11. (llm or langgraph) Add tools for deterministic lookups, retrieval, memory, review queues, or rule runners. 12. Register routers in src/api/endpoints/v1/routers.py. 13. Add tests for endpoint, usecase/action, workflow construction (langgraph only), and critical node behavior. 14. Update docs when public API shape, setup, architecture, or workflow behavior changes.

Endpoint Pattern

Endpoints should own HTTP shape only: request validation, dependency injection, background task enqueueing, response envelope, and route docs. They should not call LangGraph, tools, repositories, or model loaders directly.

For long-running jobs (requires `langgraph`), prefer:

@router.post("/thing/process", response_model=AppResponse)
async def process_thing(
    payload: ThingProcessRequest,
    background_tasks: BackgroundTasks,
    session: AsyncSession = Depends(get_db_session),
):
    usecase = ProcessThingUseCase()
    input_data = payload.model_dump()
    result = await usecase.create_job(session, input_data)
    background_tasks.add_task(usecase.process_job, result["job_id"], input_data)
    return AppResponse(success=True, data=result)

Add status and display endpoints when results are large or frontend-facing:

POST /api/v1/<feature>/process
GET  /api/v1/<feature>/status/{job_id}
GET  /api/v1/<feature>/status/{job_id}/display

Use a display endpoint to convert raw workflow output into frontend-friendly rows, summaries, totals, conflicts, held items, audit paths, and review status.

Usecase Pattern

Usecases orchestrate business flow. In production job workflows they also own:

• job creation
• status transitions
• background execution
• final status mapping from workflow output
• progress fields
• display payloads
• retry/delete/list/stats APIs when needed
• clean failure commits

Pattern:

class ProcessThingUseCase:
    def __init__(self, action: ProcessThingAction | None = None):
        self.action = action or ProcessThingAction()

    async def create_job(self, session: AsyncSession, payload: dict) -> dict:
        job = ThingJob(status="pending", input_data=payload)
        session.add(job)
        await session.commit()
        await session.refresh(job)
        return {"job_id": str(job.id), "status": "pending"}

    async def process_job(self, job_id: str, input_data: dict) -> None:
        async with async_session() as session:
            job = await self._get_job(session, job_id)
            job.status = "processing"
            await session.commit()

            try:
                result = await self.action.execute(job_id, input_data)
                final = result.get("mapping") or result.get("result") or {}
                job.status = final.get("status", "completed")
                job.result = result
                await session.commit()
            except Exception as exc:
                job.status = "failed"
                job.error = str(exc)
                await session.commit()
                # Do not re-raise after the job is recorded as failed.

Usecases should read like the business process. Move implementation details into actions, tools, repositories, providers, or utilities.

Action Pattern

Actions execute one discrete operation. For LangGraph work (requires `langgraph`), an action lazily builds and caches the workflow, then normalizes the return shape.

class ProcessThingAction:
    def __init__(self):
        self._workflow = None

    @property
    def workflow(self):
        if self._workflow is None:
            self._workflow = build_thing_workflow()
        return self._workflow

    async def execute(self, job_id: str, input_data: dict) -> dict:
        state = await self.workflow.ainvoke(
            {
                "job_id": str(job_id),
                "input_data": input_data,
            }
        )
        return {
            "job_id": str(job_id),
            "message": "Thing workflow completed",
            "result": state["final_output"],
        }

Actions should not know HTTP response details.

Workflow State Pattern

(Requires langgraph capability.)

Use TypedDict for workflow state. Keep it explicit enough that a future developer can see the system''s audit contract.

class ThingWorkflowState(TypedDict, total=False):
    job_id: str
    input_data: dict[str, Any]
    preprocess: dict[str, Any]
    ledger: dict[str, Any]
    code_rule_report: dict[str, Any]
    approved_memory_report: dict[str, Any]
    row_mappings: list[dict[str, Any]]
    coverage_report: dict[str, Any]
    conflict_report: dict[str, Any]
    summary: list[dict[str, Any]]
    totals: dict[str, Any]
    human_review_report: dict[str, Any]
    final_output: dict[str, Any]
    errors: list[dict[str, Any]]

Prefer state keys that match business reports, not internal helper names.

Workflow Build Pattern

(Requires langgraph capability.)

Keep graph factories in src/agents/workflows/<feature>/build.py. Compile and return the workflow; do not invoke it here.

def build_thing_workflow():
    workflow = StateGraph(ThingWorkflowState)
    workflow.add_node("preprocess", preprocess)
    workflow.add_node("code_rule_check", code_rule_check)
    workflow.add_node("approved_memory_check", approved_memory_check)
    workflow.add_node("reasoning", reasoning)
    workflow.add_node("coverage_validator", coverage_validator)
    workflow.add_node("conflict_check", conflict_check)
    workflow.add_node("summary", summary)
    workflow.add_node("human_review_queue", human_review_queue)
    workflow.add_node("ledger_write", ledger_write)

    workflow.add_edge(START, "preprocess")
    workflow.add_edge("preprocess", "code_rule_check")
    workflow.add_edge("code_rule_check", "approved_memory_check")
    workflow.add_edge("approved_memory_check", "reasoning")
    workflow.add_edge("reasoning", "coverage_validator")
    workflow.add_edge("coverage_validator", "conflict_check")
    workflow.add_edge("conflict_check", "summary")
    workflow.add_edge("summary", "human_review_queue")
    workflow.add_edge("human_review_queue", "ledger_write")
    workflow.add_edge("ledger_write", END)
    return workflow.compile()

Sequential edges are fine when nodes know how to no-op after a successful fast path. Use conditional edges when routing clarity is worth the extra graph structure.

Ledger And Coverage Pattern

(Requires langgraph capability.)

Use a ledger when each input item must be accounted for. This is the key production pattern for agent workflows where silent loss is unacceptable.

The ledger should track:

• batch or job id
• employee/user/entity grouping when applicable
• chunks or work units
• row/item ids
• raw source row
• normalized row
• status
• path used
• validation state
• missing, duplicate, extra, held, or conflicted items

Before final output:

• input count must equal accounted output count, or missing/held evidence must be explicit
• every mapped item has required category/result fields
• every mapped item has source identifiers and source units/facts preserved
• every fast path goes through the same coverage and conflict validators as the LLM path

Do not let an LLM output silently replace source fields such as source category, source units, source identifiers, dates, or row ids. Anchor those fields from the normalized input and only use the model for classification, reasoning, confidence, and references.

Node Pattern

(Requires langgraph capability.)

Nodes should perform one workflow step and return partial state. They may call tools, structured agents, or pure helpers, but they should not contain API or usecase logic.

LLM node shape:

async def llm_call_thing(state: ThingWorkflowState) -> ThingWorkflowState:
    context = retrieve_context(state)
    messages = [
        SystemMessage(content=prompt_manager.thing),
        HumanMessage(content=build_human_message(state, context)),
    ]
    response = await agent_manager.thing.ainvoke(messages)
    return {"thing_results": [response.model_dump()]}

Production node rules:

• Run deterministic checks before LLM calls when a safe answer may already exist.
• Preserve audit paths such as code_rule, approved_memory, pageindex, or llm.
• On invalid or incomplete model output, retry with explicit correction context.
• After max attempts, mark the chunk/item held and send it to review; do not trust partial output.
• Keep deterministic conflict checks authoritative. Optional LLM conflict checks can add evidence, not erase deterministic conflicts.

AgentManager Pattern

(Requires llm capability.)

Keep model creation and structured agents out of nodes. Use one manager that lazy-loads the model and caches agents.

class AgentManager:
    def __init__(self) -> None:
        self._model = None
        self._agents: dict[str, Any] = {}

    @property
    def model(self) -> Any:
        if self._model is None:
            self._model = get_llm("reasoning")
        return self._model

    def get_agent(self, name: str) -> Any:
        if name not in self._agents:
            schema = self._schema_for(name)
            self._agents[name] = self.model.with_structured_output(schema)
        return self._agents[name]

    @property
    def thing_reasoner(self) -> Any:
        return self.get_agent("thing_reasoner")

Use with_structured_output for normal schema-bound LLM steps. Use tool-aware agents only when the model must dynamically call deterministic tools; otherwise, call tools in the node and pass retrieved context into the prompt.

Factory Compatibility

ALMS may use function-based create_*_agent() factories alongside AgentManager. Do not replace existing factories without being asked. Use AgentManager for new production workflows; keep create_*_agent() in repos that already have them.

PromptManager Pattern

(Requires llm capability.)

Store prompts as markdown files and lazy-load them with properties. This keeps prompts editable without touching node code.

class PromptManager:
    def __init__(self, prompt_base_path: str | Path | None = None) -> None:
        self.prompt_base_path = (
            Path(prompt_base_path)
            if prompt_base_path is not None
            else Path(__file__).resolve().parent
        )
        self._thing_reasoner: str | None = None

    @property
    def thing_reasoner(self) -> str:
        if self._thing_reasoner is None:
            self._thing_reasoner = self._load_prompt("agents/agent_thing_reasoner.md")
        return self._thing_reasoner

    def _load_prompt(self, filename: str) -> str:
        return (self.prompt_base_path / filename).read_text(encoding="utf-8")

Prompts should state the output contract and safety boundaries. For source-anchored workflows, tell the model which fields are evidence and which fields it may decide.

Schema Pattern

(Requires llm capability for structured LLM outputs.)

Use Pydantic for structured LLM outputs and endpoint request models. Use TypedDict for graph state.

For production decision workflows, define separate schemas for:

• row/item mapping result
• chunk/batch mapping result
• conflict item
• conflict check result
• rule candidate specification when rule hardening exists
• human review override requests

Keep LLM output schemas strict enough to validate behavior, but not so clever that normal responses fail because of incidental formatting.

Tool Patterns

(Requires llm or langgraph capability.)

Tools are deterministic adapters around capabilities the workflow needs.

Approved memory tool:

• builds a stable exact signature from normalized source facts
• hashes the signature with sorted JSON
• looks up only approved records
• returns records keyed by signature hash
• blocks conflicting output fingerprints for the same signature

Use memory to reduce repeated LLM calls, not to generalize beyond proof.

Retrieval tool:

• hides PageIndex/vector/search implementation details
• returns scoped context for the node
• supports deterministic fallback for tests and smoke runs
• keeps retrieval config in settings

Human review queue tool (requires `database`):

• persists held chunks, conflicts, low-confidence rows, row mappings, summaries, totals, and evidence
• upserts by stable job/entity/review type when reruns happen
• returns a compact report for final output

Rule DSL runner:

• evaluates allowlisted JSON/DSL conditions
• emits mappings only when all conditions match
• can compare shadow output against trusted final mappings
• never executes arbitrary generated Python in production

Approved Memory And Rule Hardening

(Requires langgraph capability.)

Use this promotion path:

LLM/retrieval proposes mapping
-> workflow validates coverage and conflicts
-> human accepts or overrides
-> approved memory record is created
-> repeated clean memory evidence can create a shadow rule candidate
-> shadow rule is compared against trusted outputs
-> manual activation promotes it to code_rule path

Guardrails:

• Do not create memory directly from unreviewed LLM output.
• Do not fuzzy-match approved memory unless the product explicitly requires it and review accepts the risk.
• Do not activate generated rules automatically just because an LLM wrote them.
• Keep active rules all-or-nothing for a chunk or make partial fallback evidence explicit.
• Run code-rule and memory outputs through the same coverage, conflict, summary, and ledger checks as the LLM path.

Conflict And Summary Pattern

(Requires langgraph capability.)

Conflict checks should catch issues the row-level model can miss:

• held chunks/items
• missing, duplicate, or extra output rows
• low confidence
• one source category mapping to multiple target categories
• missing references or required reasoning
• total/unit reconciliation mismatch
• domain-specific policy limits

Summary aggregation should group the final row/item mappings into the shape the business needs. It should also reconcile source totals against target totals and expose documented exceptions.

When conflicts or held items exist, final status should become human_review unless the repo has a more specific status taxonomy.

Model Loader Pattern

(Requires llm capability.)

Centralize model setup through the provider abstraction. In ALMS v0.3.0 the preferred entry point is get_ai_provider():

# src/providers/ai/factory.py
from src.providers.ai.langchain_model_loader import LangchainModelLoader

def get_ai_provider() -> AIModelProvider:
    # ponytail: single provider; add google/anthropic branches here when MODEL_PROVIDER != "openai"
    return LangchainModelLoader()

# src/providers/ai/base.py
from abc import ABC, abstractmethod
from typing import Any

class AIModelProvider(ABC):
    @abstractmethod
    def get_chat_model(self, tier: str = "basic", **kwargs: Any) -> Any:
        """Return a configured chat model. tier: ''basic'' or ''reasoning''."""

# In agent or node code — vendor-neutral
from src.providers.ai.factory import get_ai_provider

model = get_ai_provider().get_chat_model("basic")
reasoning_model = get_ai_provider().get_chat_model("reasoning")

LangchainModelLoader implements AIModelProvider and supports get_chat_model(tier) — "basic" calls init_model_openai_basic, "reasoning" calls init_model_openai_reasoning. Call get_ai_provider() in lazy-loaded properties so non-AI routes do not depend on AI setup.

If the repo exposes a get_llm("reasoning") helper that wraps the same loader, prefer it for backward compatibility with existing nodes.

Google / Anthropic providers are deferred. The MODEL_PROVIDER setting is the switch point in factory.py, but only openai is implemented. Do not tell the user that Google or Anthropic providers exist unless the repo already has them.

Common settings for production agent workflows:

• AI_ENABLED — must be True to activate AI routes; gates production key validation
• DATABASE_ENABLED — controls readiness check; set False to disable DB dependency
• REDIS_ENABLED — controls Redis dependency; set False if Redis is not in use
• MODEL_PROVIDER — provider selection (currently openai only)
• OPENAI_API_KEY
• OPENAI_MODEL_BASIC
• OPENAI_MODEL_REASONING
• INFERENCE_SERVER_URL
• INFERENCE_SERVER_MODEL_BASIC
• INFERENCE_SERVER_MODEL_REASONING
• workflow max attempts
• low-confidence threshold
• unit/result reconciliation tolerance
• optional LLM conflict checker enabled flag
• raw review evidence enabled flag

Status And Display APIs

(Requires langgraph capability.)

Production workflows usually need more than one endpoint:

• process: creates the job and starts work
• status: raw lifecycle and result for polling
• display: frontend-friendly result shape
• events: optional SSE stream for live progress
• reviews: list/detail/decision for human review
• rules: build/list/activate/disable for rule hardening

The display payload should be intentionally redundant and easy for a UI to render:

job_id
status
mapping_status
path_summary
rows
summary
totals
conflicts
held_items
review
error
timestamps

Tests And Verification

Prefer focused tests by layer. Only test layers that exist in the current profile:

• endpoint test: request validation, response envelope, background enqueue/status
• usecase test: job lifecycle, status mapping, clean failure behavior
• action test: workflow invocation and result normalization (langgraph only)
• workflow test: graph compiles and minimal payload reaches final output (langgraph only)
• node test: coverage validation, conflict detection, memory hit/miss, rule match/mismatch (langgraph only)
• tool test: signature hashing, safe DSL matching, review queue persistence (llm or langgraph)
• repository test: CRUD operations (database only)

Run:

uv run pytest src/tests
uv run pytest src/tests/v1/test_<thing>.py -v
uv run ruff check src

For production agent workflows (langgraph only), add a smoke scenario that proves:

• input count equals output count
• all outputs have source ids and required result fields
• totals reconcile
• conflicts trigger review
• failed jobs are recorded cleanly
• second run can hit approved memory or active code rules only when exact coverage exists

Public Repo Notes

For a public skill or reusable starter:

• Keep generated app code free of private endpoints, API keys, internal hostnames, and domain-only secrets.
• Put domain prompts in markdown files users can swap.
• Keep examples domain-neutral unless the user asks for a specific domain.
• Document run commands using uv, because both source repos use pyproject.toml and uv.lock.
• Include minimal tests for importability and workflow construction, then add endpoint/usecase tests when behavior is stable.