Parallel Agents

@agent_service.handler()
async def run(restate_context: restate.Context, claim: InsuranceClaim) -> str:
    # Start multiple agents in parallel with auto retries and recovery
    eligibility = restate_context.service_call(run_eligibility_agent, claim)
    cost = restate_context.service_call(run_rate_comparison_agent, claim)
    fraud = restate_context.service_call(run_fraud_agent, claim)

    # Wait for all responses
    await restate.gather(eligibility, cost, fraud)

    # Run decision agent on outputs
    result = await DurableRunner.run(
        Agent(
            name="ClaimApprovalAgent", instructions="You are a claim decision engine."
        ),
        input=f"Decide about claim: {claim.model_dump_json()}. "
        "Base your decision on the following analyses:"
        f"Eligibility: {await eligibility} Cost {await cost} Fraud: {await fraud}",
    )
    return result.final_output

@agent_service.handler()
async def run(ctx: restate.ObjectContext, claim: InsuranceClaim) -> str | None:

    # Start multiple agents in parallel with auto retries and recovery
    eligibility = ctx.service_call(run_eligibility_agent, claim)
    cost = ctx.service_call(run_rate_comparison_agent, claim)
    fraud = ctx.service_call(run_fraud_agent, claim)

    # Wait for all responses
    await restate.gather(eligibility, cost, fraud)

    # Get the results
    eligibility_result = await eligibility
    cost_result = await cost
    fraud_result = await fraud

    # Run decision agent on outputs
    prompt = f"""Decide about claim: {claim.model_dump_json()}. Assessments:
    Eligibility: {eligibility_result} Cost: {cost_result} Fraud: {fraud_result}"""

    events = runner.run_async(
        user_id=ctx.key(),
        session_id=claim.session_id,
        new_message=Content(role="user", parts=[Part.from_text(text=prompt)]),
    )
    return await parse_agent_response(events)

@agent_service.handler()
async def run(ctx: restate.Context, claim: InsuranceClaim) -> str:
    # Start multiple agents in parallel with auto retries and recovery
    eligibility = ctx.service_call(run_eligibility_agent, claim)
    cost = ctx.service_call(run_rate_comparison_agent, claim)
    fraud = ctx.service_call(run_fraud_agent, claim)

    # Wait for all responses
    await restate.gather(eligibility, cost, fraud)

    # Run decision agent on outputs
    result = await restate_decision_agent.run(
        f"Decide about claim: {claim.model_dump_json()}. "
        "Base your decision on the following analyses:"
        f"Eligibility: {await eligibility} Cost {await cost} Fraud: {await fraud}",
    )
    return result.output

decision = create_agent(
    model=init_chat_model("openai:gpt-5.4"),
    system_prompt="You are a claim decision engine.",
    middleware=[RestateMiddleware()],
)


agent_service = restate.Service("ParallelAgentClaimApproval")


@agent_service.handler()
async def run(ctx: restate.Context, claim: InsuranceClaim) -> str:
    # Start multiple sub-agents in parallel with auto-retries and recovery.
    eligibility = ctx.service_call(run_eligibility_agent, claim)
    cost = ctx.service_call(run_rate_comparison_agent, claim)
    fraud = ctx.service_call(run_fraud_agent, claim)

    await restate.gather(eligibility, cost, fraud)

    result = await decision.ainvoke(
        {"messages": f"""Decide about claim: {claim.model_dump_json()}.
        Base your decision on the following analyses:
        Eligibility: {await eligibility} Cost {await cost} Fraud: {await fraud}"""}
    )
    return result["messages"][-1].content

async function run(ctx: Context, claim: ClaimInput) {
  // Create parallel tasks - each runs independently
  const claimJson = JSON.stringify(claim);
  const eligibility = ctx.run(
    "Eligibility agent",
    async () => llmCall(
        "Decide whether the following claim is eligible for reimbursement." +
        "Respond with eligible if it's a medical claim, and not eligible otherwise." +
        "\n\nClaim: " + claimJson,
    ),
    { maxRetryAttempts: 3 },
  )
  const fraud = ctx.run(
    "Fraud agent",
      async () => llmCall(
          "Decide whether the claim is fraudulent." +
          "Always respond with low risk, medium risk, or high risk." +
          "\n\nClaim: " + claimJson,
    ),
    { maxRetryAttempts: 3 },
  )
  const cost = ctx.run(
    "Rate comparison agent",
      async () => llmCall(
          "Decide whether the cost of the claim is reasonable given the treatment." +
          "Respond with reasonable or not reasonable." +
          "\n\nClaim: " + claimJson,
    ),
    { maxRetryAttempts: 3 },
  )

  // Wait for all tasks to complete and return the results
  await RestatePromise.all([eligibility, cost, fraud]);

  // Make final decision
  const { text } = await ctx.run(
      "Decision agent",
      async () => llmCall( `Decide about claim ${JSON.stringify(claim)}.
        Base your decision on the following analyses:
        - Eligibility: ${(await eligibility).text}, 
        - Cost: ${(await cost).text},
        - Fraud: ${(await fraud).text}`)
  );
  return text
}

parallelization_svc = restate.Service("ParallelAgentClaimApproval")


@parallelization_svc.handler()
async def run(ctx: restate.Context, claim: ClaimData) -> str | None:
    """Analyzes a claim in parallel with specialized agents."""

    # Create parallel tasks - each runs independently
    claim_json = json.dumps(claim.model_dump())
    eligibility = ctx.run_typed(
        "Eligibility agent",
        llm_call,
        RunOptions(max_attempts=3),
        messages="Decide whether the following claim is eligible for reimbursement."
        " Respond with eligible if it's a medical claim, and not eligible otherwise."
        f"\n\nClaim: {claim_json}",
    )
    fraud = ctx.run_typed(
        "Fraud agent",
        llm_call,
        RunOptions(max_attempts=3),
        messages="Decide whether the claim is fraudulent."
        " Always respond with low risk, medium risk, or high risk."
        f"\n\nClaim: {claim_json}",
    )
    rate = ctx.run_typed(
        "Rate comparison agent",
        llm_call,
        RunOptions(max_attempts=3),
        messages="Decide whether the cost of the claim is reasonable given the treatment."
        " Respond with reasonable or not reasonable."
        f"\n\nClaim: {claim_json}",
    )

    # Wait for all tasks to complete
    await restate.gather(eligibility, fraud, rate)

    # Make final decision
    decision = await ctx.run_typed(
        "Decision agent",
        llm_call,
        RunOptions(max_attempts=3),
        messages=f"Decide about claim: {claim.model_dump_json()}. "
        "Base your decision on the following analyses:"
        f"Eligibility: {(await eligibility).content} "
        f"Cost: {(await rate).content} "
        f"Fraud: {(await fraud).content}",
    )
    return decision.content