azure ai-foundry multi-agent claims insurance observability

基于 Microsoft Foundry 的多智能体保险理赔分诊系统

Microsoft Foundry上的保险理赔分诊系统：包括负责协调的人员、处理保险范围、欺诈问题及理赔事宜的专员、各检查点、App Insights以及容器应用托管服务。

发布者：REACTAPP.TOP加入本站 2026年9月26日阅读约 13 分钟2993 词

来源: AZURE AI: Multi-agent claims triage on Microsoft Foundry

想象一下，一家保险公司每天要通过网页表单和电话录音处理数千起新的理赔申请。在付款之前，有三项关键因素需要核查：事故发生当日的保单保障范围、SIU团队监测到的欺诈迹象，以及根据损失评估得出的理赔金额区间。人工逐一审核效率很低。而由Foundry托管的多智能体分诊系统可以让各专业人员并行工作，同时保留人工处理特殊案例。

使用的工具

Azure AI Foundry模型部署、具备任务交接功能的智能体/编排框架、用于确保持续运行的检查点机制、用于追踪流程的应用程序洞察工具，以及用于托管服务的带管理身份的容器应用。

前置条件

拥有可访问Foundry功能的Azure订阅账户、部署容器应用的能力、应用程序洞察工作空间，以及用于测试的示例理赔数据。

项目架构

一名协调员会将工作分配给三名专家（负责承保、欺诈检测和理赔）。相关工具可处理保单查询、欺诈名单核查以及分级规则的应用。检查点会保存在理赔流程的中间状态，而遥测功能则能覆盖每一次任务交接。

步骤1：创建Foundry资源并部署模型

准备好项目环境，然后部署代理将使用的聊天模型。请仔细记录这些模型的部署名称。

步骤2：安装框架并将聊天客户端指向Foundry

az group create \
  --name rg-claims-prod-uks \
  --location uksouth

az cognitiveservices account create \
  --name fdy-claims-prod-uks \
  --resource-group rg-claims-prod-uks \
  --location uksouth \
  --kind AIServices \
  --sku S0 \
  --assign-identity

az cognitiveservices account deployment create \
  --name fdy-claims-prod-uks \
  --resource-group rg-claims-prod-uks \
  --deployment-name gpt-4o \
  --model-name gpt-4o \
  --model-version 2024-08-06 \
  --model-format OpenAI \
  --sku-capacity 200 \
  --sku-name GlobalStandard

az cognitiveservices account deployment create \
  --name fdy-claims-prod-uks \
  --resource-group rg-claims-prod-uks \
  --deployment-name gpt-4o-mini \
  --model-name gpt-4o-mini \
  --model-version 2024-07-18 \
  --model-format OpenAI \
  --sku-capacity 400 \
  --sku-name GlobalStandard

python -m venv .venv
source .venv/bin/activate
pip install agent-framework azure-monitor-opentelemetry azure-identity azure-cosmos

# claims/clients.py
import os
from azure.identity import DefaultAzureCredential
from agent_framework.azure import AzureOpenAIChatClient

credential = DefaultAzureCredential()

coordinator_client = AzureOpenAIChatClient(
    endpoint=os.environ["FOUNDRY_ENDPOINT"],
    deployment_name="gpt-4o",
    credential=credential,
)

specialist_client = AzureOpenAIChatClient(
    endpoint=os.environ["FOUNDRY_ENDPOINT"],
    deployment_name="gpt-4o-mini",
    credential=credential,
)

# claims/agents.py
from claims.clients import coordinator_client, specialist_client

coordinator = coordinator_client.create_agent(
    name="triage",
    instructions=(
        "You are the first-line claims triage officer for a UK motor insurer. "
        "Read the claim and decide which specialist should handle it next. "
        "Route to 'policy_agent' when cover or policy dates are in question. "
        "Route to 'fraud_agent' when the description contains any indicator of "
        "staged loss, late notification, or inconsistent detail. "
        "Route to 'settlement_agent' when cover is confirmed and only the payout "
        "figure remains. Call the matching handoff tool, for example "
        "handoff_to_fraud_agent. Never state a settlement figure yourself."
    ),
)

policy_agent = specialist_client.create_agent(
    name="policy_agent",
    instructions=(
        "You verify whether the policy was in force on the incident date and "
        "whether the peril is covered. Use the lookup_policy tool. State the "
        "cover position plainly and cite the policy section. If cover is void, "
        "say so and stop; do not speculate on settlement."
    ),
)

fraud_agent = specialist_client.create_agent(
    name="fraud_agent",
    instructions=(
        "You score claims against the SIU signal list using the score_fraud_signals "
        "tool. Report each signal that fired and the aggregate score. Any score at "
        "or above 0.6 must be handed to policy_agent for a cover check before it "
        "goes anywhere near settlement."
    ),
)

settlement_agent = specialist_client.create_agent(
    name="settlement_agent",
    instructions=(
        "You band the damage estimate and propose a settlement using the "
        "band_settlement tool. Anything above 15000 GBP must be flagged for human "
        "authorisation in your response. Never approve a payment."
    ),
)

# claims/workflow.py
from agent_framework import HandoffBuilder
from claims.agents import coordinator, policy_agent, fraud_agent, settlement_agent

def build_triage_workflow():
    return (
        HandoffBuilder(
            name="claims_triage",
            description="Motor claims triage: cover, fraud signals, settlement banding.",
            participants=[coordinator, policy_agent, fraud_agent, settlement_agent],
        )
        .set_coordinator(coordinator)
        .add_handoff(coordinator, [policy_agent, fraud_agent, settlement_agent])
        .add_handoff(fraud_agent, policy_agent)
        .add_handoff(policy_agent, settlement_agent)
        .with_interaction_mode("autonomous", autonomous_turn_limit=12)
        .with_termination_condition(
            lambda conv: any(
                "AUTHORISATION REQUIRED" in (msg.text or "")
                for msg in conv[-2:]
            )
        )
        .build()
    )

# claims/tools.py
from typing import Annotated
from datetime import date
import httpx

POLICY_API = "https://pas.internal.claims-leeds.co.uk/v2"

async def lookup_policy(
    policy_number: Annotated[str, "The policy number, format MOT-00000000"],
    incident_date: Annotated[str, "Incident date as YYYY-MM-DD"],
) -> dict:
    """Return cover status and policy sections in force on the incident date."""
    async with httpx.AsyncClient(timeout=10.0) as client:
        r = await client.get(
            f"{POLICY_API}/policies/{policy_number}/cover",
            params={"as_at": incident_date},
        )
        r.raise_for_status()
        return r.json()

SIU_SIGNALS = {
    "late_notification_days": 14,
    "no_police_reference": True,
    "prior_claims_12m": 2,
}

def score_fraud_signals(
    description: Annotated[str, "Free-text claim description"],
    notified_after_days: Annotated[int, "Days between incident and notification"],
    prior_claims_12m: Annotated[int, "Claims by this policyholder in last 12 months"],
) -> dict:
    """Score a claim against the SIU signal list. Returns fired signals and aggregate."""
    fired = []
    if notified_after_days > SIU_SIGNALS["late_notification_days"]:
        fired.append("late_notification")
    if prior_claims_12m >= SIU_SIGNALS["prior_claims_12m"]:
        fired.append("claim_frequency")
    if "no witnesses" in description.lower():
        fired.append("no_independent_witness")
    return {"signals_fired": fired, "score": round(len(fired) / 3, 2)}

def band_settlement(
    estimate_gbp: Annotated[float, "Repair or replacement estimate in GBP"],
) -> dict:
    """Band an estimate and return the proposed settlement and authorisation route."""
    if estimate_gbp <= 2500:
        return {"band": "A", "proposed_gbp": estimate_gbp, "route": "auto"}
    if estimate_gbp <= 15000:
        return {"band": "B", "proposed_gbp": estimate_gbp, "route": "team_leader"}
    return {"band": "C", "proposed_gbp": estimate_gbp, "route": "AUTHORISATION REQUIRED"}

policy_agent = specialist_client.create_agent(
    name="policy_agent",
    instructions="...",
    tools=[lookup_policy],
)

fraud_agent = specialist_client.create_agent(
    name="fraud_agent",
    instructions="...",
    tools=[score_fraud_signals],
)

settlement_agent = specialist_client.create_agent(
    name="settlement_agent",
    instructions="...",
    tools=[band_settlement],
)

# claims/checkpoints.py
from agent_framework import InMemoryCheckpointStorage

# Tests and local runs
storage = InMemoryCheckpointStorage()

workflow = build_triage_workflow_with_storage(storage)

async for event in workflow.run_stream(claim_text, session_id=claim_id):
    handle(event)

# claims/telemetry.py
import os
from azure.monitor.opentelemetry import configure_azure_monitor
from agent_framework.observability import enable_instrumentation

def init_telemetry() -> None:
    configure_azure_monitor(
        connection_string=os.environ["APPLICATIONINSIGHTS_CONNECTION_STRING"],
    )
    enable_instrumentation()

dependencies
| where timestamp > ago(24h)
| where name startswith "invoke_agent"
| extend agent = tostring(customDimensions["gen_ai.agent.name"])
| extend claim = tostring(customDimensions["session.id"])
| summarize
    hops = count(),
    total_ms = sum(duration),
    p95_ms = percentile(duration, 95)
    by agent
| order by total_ms desc

# claims/run.py
import asyncio
from agent_framework import RequestInfoEvent
from claims.telemetry import init_telemetry
from claims.workflow import build_triage_workflow

CLAIM = """
Claim MOT-40118822. Policy MOT-00318844, incident 2026-07-29, notified 2026-08-19.
Insured reports offside wing and door damage in a supermarket car park, no witnesses,
no police reference. Third party details not supplied. Estimate 3,180 GBP.
"""

async def main() -> None:
    init_telemetry()
    workflow = build_triage_workflow()

    async for event in workflow.run_stream(CLAIM, session_id="MOT-40118822"):
        if isinstance(event, RequestInfoEvent):
            # Autonomous mode only surfaces this on escalation
            print(f"HUMAN NEEDED: {event.data.prompt}")
            break
        print(event)

asyncio.run(main())

az identity create \
  --name umi-claims-triage \
  --resource-group rg-claims-prod-uks

PRINCIPAL=$(az identity show -n umi-claims-triage -g rg-claims-prod-uks --query principalId -o tsv)
CLIENT=$(az identity show -n umi-claims-triage -g rg-claims-prod-uks --query clientId -o tsv)
FOUNDRY=$(az cognitiveservices account show -n fdy-claims-prod-uks -g rg-claims-prod-uks --query id -o tsv)

az role assignment create \
  --role "Azure AI User" \
  --assignee-object-id $PRINCIPAL \
  --assignee-principal-type ServicePrincipal \
  --scope $FOUNDRY

az containerapp create \
  --name ca-claims-triage \
  --resource-group rg-claims-prod-uks \
  --environment cae-claims-prod-uks \
  --image acrclaimsproduks.azurecr.io/claims-triage:1.0.0 \
  --user-assigned umi-claims-triage \
  --min-replicas 1 \
  --max-replicas 8 \
  --cpu 1.0 \
  --memory 2.0Gi \
  --env-vars \
    FOUNDRY_ENDPOINT=https://fdy-claims-prod-uks.openai.azure.com/ \
    AZURE_CLIENT_ID=$CLIENT \
    ENABLE_INSTRUMENTATION=true \
    OTEL_SERVICE_NAME=claims-triage \
    APPLICATIONINSIGHTS_CONNECTION_STRING=secretref:appi-conn

az group delete \
  --name rg-claims-prod-uks \
  --yes \
  --no-wait

az containerapp delete \
  --name ca-claims-triage \
  --resource-group rg-claims-prod-uks \
  --yes

az cognitiveservices account deployment delete \
  --name fdy-claims-prod-uks \
  --resource-group rg-claims-prod-uks \
  --deployment-name gpt-4o-mini

配置端点、部署信息以及Entra/托管身份认证机制——切勿在容器镜像中存放API密钥。

步骤3：确定协调员与三名专家的角色

为每位专家设定明确的系统提示和工具集。协调员仅负责任务分配与信息汇总，不得重复实现专家的逻辑功能。

步骤4：构建任务传递关系图

明确的任务传递路径比混乱的工具组合更有效。默认流程应为“风险识别→欺诈检测→结算处理”；当欺诈评分异常升高时，可直接转交至专项调查组。

步骤5：为专家提供真实工具

这些工具必须能够连接实际系统，或是具备完整结构定义的模拟工具。那些总是“批准”请求的虚假工具毫无教学价值。

步骤6：设置检查点以防止任务中断后数据丢失

在每位专家处理完任务后都要保存相应的线程/请求状态。即使在处理过程中容器重启，也应能继续处理未完成的任务，而非重复支付。

步骤7：将追踪数据与指标发送至Application Insights

为任务交接、工具延迟以及最终处理结果触发自定义事件。控制面板应显示与手动基准值相比的中间值处理时间。

第8步：运行索赔请求并查看结果

使用已知承保范围和欺诈检测结果的示例索赔请求。确保协调员生成的最终报告包含所有三项专家分析结果。

第9步：在具有托管身份的容器应用中部署

为Foundry部署以及工具调用的所有后端API授予相应身份权限。严格限制外部访问。

故障排除

专家分析结果为空通常意味着部署名称有误或工具的JSON数据被截断。出现循环情况往往是因为缺少任务交接的终止条件。认证失败是由于角色权限配置问题，而非“模型故障”。

运行此系统的成本

模型令牌占主导地位；检查点存储和 App Insights 的占用量相对较小。需限制每条请求的最大处理轮数以及缓存策略查询次数。

清理工作

删除为演示流程创建的容器应用、身份凭证及部署配置，以避免相关资源持续计入账单。

总结

当专家领域较为明确、工具真实可用、检查点持久稳定，且遥测数据能证明延迟对性能的影响时，多智能体请求分类机制才能发挥作用。Foundry 负责提供模型基础框架，而图结构与操作流程则构成最终产品。

对于超过阈值的情况，应增设人工审批环节。智能体应加快分类处理速度，而非擅自批准大额赔付。

将版本欺诈检测规则与模型提示词分开管理。当 SIU 更新信号列表时，无需修改提示词即可自动识别新信号——相关工具应能直接读取当前列表内容。

对Foundry部署环境中的并发请求进行压力测试，以验证其是否符合速率限制要求。必须明确界定队列溢出时的处理方式：拒绝请求、延迟处理或转为人工处理。

隐私保护：索赔相关文本属于敏感信息。需确保日志中不会泄露个人身份识别号码或社会安全号码，并且App Insights的采样功能不得在自定义维度中存储原始个人身份信息。

需设计一种混沌测试，能够在索赔处理过程中终止容器运行，并验证系统能否从检查点恢复且不会产生重复的工具副作用——写入操作时必须使用幂等性键。