Skip to Content
ExamplesPythonMulti-Turn Workflow

Multi-Turn Workflow

Orchestrate three sequential Pydantic AI agents — a classifier, a risk assessor, and a decision agent — under a single VeriProof session. Each agent run is a separate step with its own governance metadata, but all steps share a single Merkle-sealed audit trail.

PythonMulti-agentPython ≥ 3.10

Prerequisites

pip install veriproof-sdk veriproof-sdk-instrumentation-pydantic-ai pydantic-ai

Environment

VERIPROOF_API_KEY=vp_cust_your-tenant.<key-material>
VERIPROOF_APPLICATION_ID=loan-workflow
ANTHROPIC_API_KEY=sk-ant-...

Complete example

import asyncio
import os
from veriproof_sdk import (
    configure_veriproof,
    VeriproofClientOptions,
    AgentRole,
    DecisionType,
    RiskLevel,
    SessionIntent,
)
from veriproof_sdk_instrumentation_pydantic_ai import VeriproofPydanticAISession
from pydantic_ai import Agent
from pydantic_ai.models.anthropic import AnthropicModel
 
configure_veriproof(
    VeriproofClientOptions(
        api_key=os.environ["VERIPROOF_API_KEY"],
        application_id=os.environ["VERIPROOF_APPLICATION_ID"],
    ),
    service_name=os.environ["VERIPROOF_APPLICATION_ID"],
    set_global=True,
)
 
# ── Agent definitions ──────────────────────────────────────────────────────
 
classify_agent = Agent(
    AnthropicModel("claude-3-5-haiku-20241022"),
    system_prompt=(
        "Classify the loan application. "
        "Return one of: STANDARD | COMPLEX | HIGH_RISK. "
        "Respond with the classification only."
    ),
)
 
risk_agent = Agent(
    AnthropicModel("claude-3-5-haiku-20241022"),
    system_prompt=(
        "You are a credit risk analyst. "
        "Given a loan classification and financials, produce a risk score (0–100) "
        "and list up to 3 risk factors. Format: SCORE:<n> FACTORS:<factor1>|<factor2>"
    ),
)
 
decision_agent = Agent(
    AnthropicModel("claude-3-5-sonnet-20241022"),
    system_prompt=(
        "You are a senior loan officer. "
        "Based on the classification and risk assessment, make a final APPROVE or DECLINE decision. "
        "State the reason in one sentence."
    ),
)
 
 
async def process_loan_application(applicant_id: str, credit_score: int, income: float) -> dict:
    """Run the full three-step loan decision workflow under a single VeriProof session."""
 
    application_summary = (
        f"Applicant: {applicant_id}. "
        f"Credit score: {credit_score}. "
        f"Annual income: ${income:,.0f}. "
        f"Requested loan: $25,000. Debt-to-income ratio: 0.34."
    )
 
    # Create one session shared across all three agent runs.
    # All agent_run spans carry the same veriproof.session.id.
    session = VeriproofPydanticAISession(
        session_intent=SessionIntent.ADVISORY,
        decision_type=DecisionType.APPROVAL,
        agent_role=AgentRole.SPECIALIST,
    )
 
    # ── Step 1: Classify ──────────────────────────────────────────────
    classify_result = await session.run(classify_agent, application_summary)
    classification = classify_result.data.strip()
 
    # ── Step 2: Risk assessment ───────────────────────────────────────
    risk_prompt = f"{application_summary} Classification: {classification}."
    risk_result = await session.run(risk_agent, risk_prompt)
    risk_text = risk_result.data
 
    # Parse score from response (e.g. "SCORE:42 FACTORS:high-dti|limited-history")
    risk_score = 50  # default
    if "SCORE:" in risk_text:
        try:
            risk_score = int(risk_text.split("SCORE:")[1].split()[0])
        except (ValueError, IndexError):
            pass
 
    risk_level = (
        RiskLevel.HIGH if risk_score >= 70
        else RiskLevel.MEDIUM if risk_score >= 40
        else RiskLevel.LOW
    )
    session.set_risk_level(risk_level.value)
 
    # ── Step 3: Final decision ────────────────────────────────────────
    session.set_decision_type(
        DecisionType.APPROVAL if risk_score < 70 else DecisionType.DENIAL
    )
    session.set_agent_role(AgentRole.PRIMARY)
 
    decision_prompt = (
        f"{application_summary} "
        f"Classification: {classification}. Risk score: {risk_score}/100."
    )
    decision_result = await session.run(decision_agent, decision_prompt)
    decision_text = decision_result.data
    approved = "APPROVE" in decision_text.upper()
 
    return {
        "session_id": session.session_id,
        "classification": classification,
        "risk_score": risk_score,
        "approved": approved,
        "decision": decision_text,
    }
 
 
if __name__ == "__main__":
    result = asyncio.run(
        process_loan_application("A-1042", credit_score=712, income=82000)
    )
    print(f"Session   : {result['session_id']}")
    print(f"Class     : {result['classification']}")
    print(f"Risk      : {result['risk_score']}/100")
    print(f"Decision  : {'APPROVED' if result['approved'] else 'DECLINED'}")
    print(f"Reasoning : {result['decision']}")

What you’ll see in VeriProof

All three agent runs appear under the same veriproof.session.id:

agent_run pydantic-ai   (classify_agent — session.session_id)
agent_run pydantic-ai   (risk_agent     — same session_id)
agent_run pydantic-ai   (decision_agent — same session_id, risk_level=high/medium/low)

Each agent_run span carries veriproof.session.id, veriproof.session.intent=advisory, and the governance attributes set at the time of that run.

The three-agent pattern above is a simple sequential pipeline. For branching or parallel execution, use LangGraph to model your graph and the LangGraph instrumentation adapter to capture it.

Adding a human review gate

If your compliance requirements mandate human sign-off before actioning the decision, record the oversight outcome using the session’s fluent setters before the final agent run:

# Signal that human approval is required for this decision
session.set_decision_type(DecisionType.APPROVAL)
session.with_metadata("veriproof.human.oversight", "approval_required")
session.with_metadata("veriproof.human.reviewer_id", "reviewer:R-007")

This adds Article 14 (transparency) metadata to the session evidence record in VeriProof’s EU AI Act compliance export.

Next steps

Last updated on
LangGraph ChatbotSemantic Kernel