AI Agents

Designing a Multi-Agent Field Service Assistant

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

Field technicians don’t have time to fight software. They need a concise briefing before a job and a fast, hands-free way to log what happened after. We built a multi-agent conversational assistant for exactly that — Python, FastAPI, and Pydantic v2 — and this post walks through how it’s structured.

Why multiple agents

A single monolithic prompt trying to brief, debrief, and classify would be brittle and impossible to evaluate. Instead, each responsibility is its own agent with a narrow contract, coordinated by an orchestrator:

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, cheaper to test, and independently evaluable — which matters a lot once you start gating deploys on evaluation results.

The request flows top-down: the technician talks to the FastAPI surface, the orchestrator drives the three agents, and those 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

Generates a pre-job briefing before the technician arrives. It performs parallel retrieval across the equipment, contract, IoT, and work-order connectors, then produces three output formats from one payload:

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

Producing all three from a single retrieval pass keeps latency down and the content consistent across channels.

Debriefing agent

Runs an 8-step post-job debrief as a guided 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

Structuring the debrief as explicit steps means the agent always knows where it is, and the evaluation harness can verify each step produced the right system state.

Those steps are modeled as enums, and the collected data is a Pydantic model — so “where are we” and “what have we gathered” are both typed, not free-floating dict 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 drives the state machine defensively — every turn works on a deep copy of the session, and the mutation is only committed once persistence succeeds. That’s what makes a dropped WebSocket or a mid-turn crash non-destructive:

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

Classifies service activities using keyword rules first, falling back to an LLM only for ambiguous cases, and returns a category, subcategory, confidence, and reasoning. (It’s worth its own post — see hybrid activity classification.)

Connectors: real system, swappable backends

Every external dependency sits behind an abstract connector with a mock implementation. A blank endpoint in config selects the mock; a real 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}

This is what lets the whole assistant run locally with no Azure services for development and tests, while the same code path talks to Azure OpenAI and Azure AI Search in production. Tests exercise the real orchestration, not a pile of mocks that drift from reality.

The feature knobs that shape behavior are config too, so the same build behaves differently per deployment:

Persistence and readiness

The assistant owns two Cosmos DB containers — debriefs (partition key /session_id) and service-reports (partition key /report_id). A deliberate design rule: runtime stores never create containers. They’re provisioned once, up front, by an idempotent seeding step or by the infra Bicep modules. When AZURE_COSMOS_ENDPOINT is unset, persistence falls back to a durable local JSON file store.

Health endpoints reflect that split:

• GET /health — liveness only; 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 keeps orchestrators from killing a healthy process just because a downstream store is warming up.

Outcome verification, not vibes

Because the debrief is typed, “did the agent actually finish the job” is a checkable predicate, not a judgment call. A small set of required fields must hold real values before a session can be called complete:

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

The validation is per-field and strict — steps_taken isn’t “present”, it’s “a non-empty list of records each with a non-empty action_description”. That precision is what lets the evaluation harness assert on the resulting system state instead of on the transcript, and it’s the same idea the evaluation series leans on hard.

The API surface

Takeaways

• Split by responsibility. Narrow agents are easier to prompt, test, and evaluate than one mega-prompt.
• Hide backends behind connectors so the system runs locally and in the cloud from the same code.
• Provision persistence explicitly and separate liveness from readiness.

The payoff of this structure shows up when you evaluate it — which is the subject of a whole series of its own.

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