AI Agents

Designing a Multi-Agent Field Service Assistant

By Omer DemirMar 10, 2026⏱ 6 min read👁 — reads⧉ — copies

AI disclosure: This article was moderated by AI.

Field technicians need a concise briefing before a job and a fast, hands-free way to record what happened afterward. We built a multi-agent conversational assistant for that workflow using Python, FastAPI, and Pydantic v2. Its architecture separates briefing, debriefing, and classification work behind an orchestrator.

Why multiple agents

A single prompt responsible for briefing, debriefing, and classification would be brittle and impossible to evaluate. Each responsibility therefore has its own agent and narrow contract, with an orchestrator coordinating the work:

src/
├── main.py                 # FastAPI entry point
├── agents/
│   ├── briefing/           # Pre-job briefing agent
│   ├── debriefing/         # 8-step post-job debrief agent
│   ├── classification/     # Keyword-rule + LLM activity classifier
│   └── orchestrator/       # Session state machine
├── connectors/             # Abstract + mock data connectors
├── models/                 # Pydantic v2 domain models
└── prompts/                # LLM prompt templates (.prompt.md)

Narrow agents are easier to prompt and cheaper to test. They can also be evaluated independently, which matters once evaluation results gate deployments.

Requests flow top-down. The technician talks to the FastAPI surface, the orchestrator drives the three agents, and the agents reach connectors, Azure OpenAI / Voice Live, and persistence. Source: diagrams/multi-agent-architecture.drawio (editable in the draw.io VS Code extension).

The agents

Briefing agent

Before the technician arrives, the briefing agent retrieves data in parallel from the equipment, contract, IoT, and work-order connectors. It produces three formats from the same payload:

• Voice is limited to ≤200 words for text-to-speech in the vehicle.
• Mobile provides a compact display format.
• Markdown provides full detail for reference.

One retrieval pass keeps latency down and makes the content consistent across channels.

Debriefing agent

The debriefing agent runs an 8-step post-job conversation:

1. Greet & confirm job completion
2. Fault observed on arrival
3. Steps taken (diagnosis & repair)
4. Parts replaced
5. Tests performed
6. Operational status
7. Image documentation
8. Complete & close

Explicit steps give the agent a known position in the conversation and let the evaluation harness verify that each step produced the expected system state.

Enums model the steps, while a Pydantic model holds the collected data. Both the current position and the gathered information are typed instead of stored in free-floating dictionary keys:

class SessionState(str, Enum):
    STARTED = "started"; IN_PROGRESS = "in_progress"
    REPORT_COLLECTED = "report_collected"; COMPLETE = "complete"

class DebriefStep(str, Enum):
    GREET_AND_CONFIRM = "greet_and_confirm"; FAULT_OBSERVED = "fault_observed"
    STEPS_TAKEN = "steps_taken"; PARTS_REPLACED = "parts_replaced"
    TESTS_PERFORMED = "tests_performed"; OPERATIONAL_STATUS = "operational_status"
    IMAGE_DOCUMENTATION = "image_documentation"; COMPLETE = "complete"

class DebriefingData(BaseModel):
    fault_observed: str | None = None
    steps_taken: list[ActivityRecord] = Field(default_factory=list)
    parts_replaced: list[PartRecord] = Field(default_factory=list)
    tests_performed: list[str] = Field(default_factory=list)
    operational_status: OperationalStatus | None = None
    images: list[str] = Field(default_factory=list)

The orchestrator handles each turn on a deep copy of the session. It commits the mutation only after persistence succeeds, so a dropped WebSocket or mid-turn crash does not damage the live state:

async def process_message(self, session_id, message):
    debrief_session = self._sessions[session_id]
    candidate = debrief_session.model_copy(deep=True)   # never mutate live state

    response = await self._debriefing_agent.run(message, debrief_session=candidate)

    if candidate.state in {SessionState.REPORT_COLLECTED, SessionState.COMPLETE}:
        candidate = await self._prepare_collection_candidate(candidate)
    await self._persist_and_publish(candidate)          # commit only after success
    return candidate, response.text

Activity classifier

The classifier applies keyword rules first and calls an LLM only for ambiguous cases. It returns a category, subcategory, confidence, and reasoning. The hybrid activity classification article covers that design in detail.

Swappable connector backends

Every external dependency sits behind an abstract connector with a mock implementation. A blank endpoint selects the mock, while a configured endpoint selects the live client:

providers:
  llm:
    type: ${LLM_PROVIDER:azure_openai}
    endpoint: ${AZURE_OPENAI_ENDPOINT}
    deployment: ${AZURE_OPENAI_DEPLOYMENT:gpt-4.1}
  search:
    # Blank endpoint selects the local mock retrieval connector.
    endpoint: ${AZURE_SEARCH_ENDPOINT}

The assistant can run locally without Azure services for development and tests. In production, the same code path talks to Azure OpenAI and Azure AI Search. Tests still exercise the real orchestration rather than a separate set of mocks that can drift from production behavior.

Behavioral settings also live in configuration, allowing the same build to serve different deployments:

Persistence and readiness

The assistant owns two Cosmos DB containers: debriefs with partition key /session_id, and service-reports with partition key /report_id. Runtime stores never create containers. An idempotent seeding step or the infrastructure Bicep modules provision them in advance. When AZURE_COSMOS_ENDPOINT is unset, persistence falls back to a durable local JSON file store.

The health endpoints follow the same division:

• GET /health checks liveness only and never touches Cosmos.
• GET /health/ready checks the selected persistence mode. In Cosmos mode, it returns ready only after both owned containers are readable.

Separating liveness from readiness prevents orchestrators from killing a healthy process while a downstream store warms up.

Verifying outcomes

Because the debrief is typed, completion is a checkable predicate. A small set of required fields must contain real values before the session can be marked complete:

REQUIRED_DEBRIEF_FIELDS = (
    "fault_observed", "steps_taken", "tests_performed", "operational_status",
)

Validation is strict for each field. For example, steps_taken must be a non-empty list of records whose action_description values are also non-empty. This precision lets the evaluation harness assert against the resulting system state instead of the transcript. The evaluation series applies the same principle.

The API surface

Why the structure holds up

Splitting responsibilities produces agents that are easier to prompt, test, and evaluate than one large prompt. Connectors let the same code run locally and in the cloud, while explicit provisioning and separate liveness and readiness checks keep persistence behavior predictable.

The evaluation series shows how this structure supports deployment gates and repeatable agent testing.

#multi-agent#fastapi#pydantic#azure-openai#architecture