Human-in-the-loop¶

import os

# Get the API base, API key and model name.
_api_key = os.environ.get("OPENAI_API_KEY")
_api_base = os.environ.get("OPENAI_API_BASE")
_model_name = os.environ.get("OPENAI_MODEL_NAME")

from pydantic import BaseModel, Field
from bridgic.core.automa import GraphAutoma, worker, Snapshot
from bridgic.core.automa.args import From
from bridgic.core.automa.interaction import Event, Feedback, FeedbackSender, InteractionFeedback, InteractionException
from bridgic.core.model.types import Message, Role
from bridgic.core.model.protocols import PydanticModel
from bridgic.llms.openai import OpenAILlm

# Set the LLM
llm = OpenAILlm(api_base=_api_base, api_key=_api_key, timeout=10)

class CodeBlock(BaseModel):
    code: str = Field(description="The code to be executed.")

class CodeAssistant(GraphAutoma):
    @worker(is_start=True)
    async def generate_code(self, user_requirement: str):
        response = await llm.astructured_output(
            model=_model_name,
            messages=[
                Message.from_text(text=f"You are a programming assistant. Please generate code according to the user's requirements.", role=Role.SYSTEM),
                Message.from_text(text=user_requirement, role=Role.USER),
            ],
            constraint=PydanticModel(model=CodeBlock)
        )
        return response.code

    @worker(dependencies=["generate_code"])
    async def ask_to_run_code(self, code: str):
        event = Event(event_type="can_run_code", data=code)
        feedback = await self.request_feedback_async(event)
        return feedback.data
        
    @worker(dependencies=["ask_to_run_code"])
    async def output_result(self, feedback: str, code: str = From("generate_code")):
        code = code.strip("```python").strip("```")
        if feedback == "yes":
            print(f"- - - - - - Result - - - - - -")
            exec(code)
            print(f"- - - - - - End - - - - - -")
        else:
            print(f"This code was rejected for execution. In response to the requirements, I have generated the following code:\n```python\n{code}\n```")

# Handle can_run_code event
def can_run_code_handler(event: Event, feedback_sender: FeedbackSender):
    print(f"Can I run this code now to verify if it's correct?")
    print(f"```python\n{event.data}\n```")
    res = input("Please input your answer (yes/no): ")
    if res in ["yes", "no"]:
        feedback_sender.send(Feedback(data=res))
    else:
        print("Invalid input. Please input yes or no.")
        feedback_sender.send(Feedback(data="no"))

# register can_run_code event handler to `CodeAssistant` automa
code_assistant = CodeAssistant()
code_assistant.register_event_handler("can_run_code", can_run_code_handler)

import os
import tempfile
from httpx import delete
from pydantic import BaseModel
from datetime import datetime
from bridgic.core.automa import GraphAutoma, worker, Snapshot
from bridgic.core.automa.args import From
from bridgic.core.automa.interaction import Event, InteractionFeedback, InteractionException

class ReimbursementRecord(BaseModel):
    request_id: int
    employee_id: int
    employee_name: str
    reimbursement_month: str
    reimbursement_amount: float
    description: str
    created_at: datetime
    updated_at: datetime

class AuditResult(BaseModel):
    request_id: int
    passed: bool
    audit_reason: str

class ReimbursementWorkflow(GraphAutoma):
    @worker(is_start=True)
    async def load_record(self, request_id: int):
        """
        The reimbursement workflow can be triggered by the OA system — for instance, when an employee submits a new reimbursement request. Each request is uniquely identified by a `request_id`, which is then used to retrieve the corresponding reimbursement record from the database. 
        """
        # Load the data from database.
        return await self.load_record_from_database(request_id)
    
    @worker(dependencies=["load_record"])
    async def audit_by_rules(self, record: ReimbursementRecord):
        """
        This method simulates the logic for automatically determining whether a reimbursement request complies with business rules.  

        Typical reasons for a reimbursement request failing the audit include:

        - Unusually large individual amounts
        - Excessive total amounts within a month
        - Expenses that do not meet reimbursement policies
        - Missing or invalid supporting documents
        - Duplicate submissions
        - Other non-compliant cases
        """
        if record.reimbursement_amount > 2500:
            return AuditResult(
                request_id=record.request_id,
                passed=False,
                audit_reason="The reimbursement amount {record.reimbursement_amount} exceeds the limit of 2500."
            )
        # TODO: Add more audit rules here.
        ...

        return AuditResult(
            request_id=record.request_id,
            passed=True,
            audit_reason="The reimbursement request passed the audit."
        )
    
    @worker(dependencies=["audit_by_rules"])
    async def execute_payment(self, result: AuditResult, record: ReimbursementRecord = From("load_record")):
        if not result.passed:
            print(f"The reimbursement request {record.request_id} failed the audit. Reason: {result.audit_reason}")
            return False
        
        # The reimbursement request {record.request_id} has passed the audit rules. Requesting approval from the manager...
        # human-in-the-loop: request approval from the manager.
        event = Event(
            event_type="request_approval",
            data={
                "reimbursement_record": record,
                "audit_result": result
            }        
        )
        feedback: InteractionFeedback = self.interact_with_human(event)
        if feedback.data == "yes":
            await self.lanuch_payment_transaction(record.request_id)
            print(f"The reimbursement request {record.request_id} has been approved. Payment transaction launched.")
            return True

        print(f"!!!The reimbursement request {record.request_id} has been rejected. Payment transaction not launched.\nRejection info:\n {feedback.data}")
        return False

    async def load_record_from_database(self, request_id: int):
        # Simulate a database query...
        return ReimbursementRecord(
            request_id=request_id,
            employee_id=888888,
            employee_name="John Doe",
            reimbursement_month="2025-10",
            reimbursement_amount=1024.00,
            description="Hotel expenses for a business trip",
            created_at=datetime(2025, 10, 11, 10, 0, 0),
            updated_at=datetime(2025, 10, 11, 10, 0, 0)
        )
    async def lanuch_payment_transaction(self, request_id: int):
        # Simulate a payment execution...
        ...