AgentOps-AI 实战：基于 Google ADK 实现人工审批工作流

概述

在现代企业自动化流程中，人工审批环节往往是不可或缺的关键节点。本文将介绍如何利用 AgentOps-AI 项目结合 Google ADK（Agent Development Kit）构建一个带有人工审批环节的智能工作流系统。这个系统能够自动处理审批请求的预处理，在关键节点引入人工决策，并最终完成整个审批流程。

技术架构

该工作流系统主要由以下几个核心组件构成：

1. Google ADK：提供基础的 Agent 开发框架
2. AgentOps：用于工作流的可观测性追踪
3. 多智能体协作：多个 Agent 协同完成不同阶段的任务
4. 人工干预机制：在关键决策点引入人类审批

环境准备

首先需要安装必要的依赖包：

!pip install google-adk agentops python-dotenv nest_asyncio asyncio

然后导入所需的库：

import json
import os
import asyncio
from google.adk.agents import LlmAgent, SequentialAgent
from google.adk.tools import FunctionTool
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from pydantic import BaseModel, Field
import nest_asyncio
import agentops
from dotenv import load_dotenv

数据模型设计

我们使用 Pydantic 定义审批请求和审批决策的数据结构：

class ApprovalRequest(BaseModel):
    amount: float = Field(description="需要审批的金额")
    reason: str = Field(description="审批原因")

class ApprovalDecision(BaseModel):
    decision: str = Field(description="审批结果：'approved' 或 'rejected'")
    comments: str = Field(description="审批人的附加说明")

这种强类型定义确保了数据的一致性和可验证性。

人工审批工具实现

核心的人工审批工具实现如下：

async def external_approval_tool(amount: float, reason: str) -> str:
    """ 
    向用户请求人工审批并返回决策结果(JSON格式)
    """
    print(f"🔔 需要人工审批:")
    print(f"   金额: ${amount:,.2f}")
    print(f"   原因: {reason}")
    decision = ""
    while decision.lower() not in ["approved", "rejected"]:
        decision = input("   请输入审批决定(approved/rejected): ").strip().lower()
        if decision.lower() not in ["approved", "rejected"]:
            print("   输入无效，请输入 'approved' 或 'rejected'")
    comments = input("   请输入审批意见(可选): ").strip()
    print(f"   决定: {decision.upper()}")
    print(f"   意见: {comments if comments else '无'}")
    return json.dumps({
        "decision": decision,
        "comments": comments,
        "amount": amount,
        "reason": reason
    })

这个工具会：

1. 显示需要审批的金额和原因
2. 等待用户输入审批决定
3. 可选的审批意见
4. 返回结构化的审批结果

多智能体工作流设计

我们设计了三个专门的 Agent 来完成审批流程：

1. 审批请求准备 Agent

prepare_request = LlmAgent(
    model=MODEL_NAME,
    name="PrepareApprovalAgent",
    description="从用户输入中提取并准备审批请求详情",
    instruction="""你是一个审批请求准备Agent。
        你的任务：
        1. 从用户请求中提取金额和原因
        2. 将这些值存储在会话状态中，键为'approval_amount'和'approval_reason'
        3. 验证是否提供了金额和原因
        4. 响应将要提交审批的内容摘要
        
        如果用户输入缺少金额或原因，请要求澄清。
    """,
    output_key="request_prepared"
)

2. 人工审批请求 Agent

request_approval = LlmAgent(
    model=MODEL_NAME,
    name="RequestHumanApprovalAgent", 
    description="使用外部审批系统处理准备好的请求详情",
    instruction="""你是一个人工审批请求Agent。
        你的任务：
        1. 从会话状态获取'approval_amount'和'approval_reason'
        2. 使用external_approval_tool处理这些值
        3. 将审批决定存储在会话状态中，键为'human_decision'
        4. 响应审批状态
        
        始终使用会话状态中的确切值进行工具调用。
    """,
    tools=[approval_tool],
    output_key="approval_requested"
)

3. 决策处理 Agent

process_decision = LlmAgent(
    model=MODEL_NAME,
    name="ProcessDecisionAgent",
    description="处理人工审批决定并提供最终响应",
    instruction="""你是一个决策处理Agent。
        你的任务：
        1. 检查会话状态中的'human_decision'
        2. 解析审批决策JSON
        3. 如果批准：祝贺并提供后续步骤
        4. 如果拒绝：解释拒绝原因并建议替代方案
        5. 向用户提供清晰、有帮助的最终响应

        在响应中保持专业和帮助的态度。
    """,
    output_key="final_decision"
)

工作流编排

将上述 Agent 组合成顺序工作流：

approval_workflow = SequentialAgent(
    name="HumanApprovalWorkflowNotebook",
    description="带有人工监督的完整审批请求处理工作流",
    sub_agents=[prepare_request, request_approval, process_decision]
)

运行工作流

定义工作流运行函数：

async def run_approval_workflow_notebook(user_request: str, session_id: str):
    """在笔记本环境中运行完整的审批工作流"""
    print(f"{'='*60}")
    print(f" 开始审批工作流 - 会话ID: {session_id}")
    print(f"{'='*60}")
    print(f"用户请求: {user_request}")
    
    user_content = types.Content(
        role='user', 
        parts=[types.Part(text=user_request)]
    )
    
    step_count = 0
    final_response = "未收到响应"
    
    async for event in workflow_runner.run_async(
        user_id=USER_ID,
        session_id=session_id,
        new_message=user_content,
    ):
        if event.author and event.content:
            step_count += 1
            print(f"📋 步骤 {step_count} - {event.author}:")
            if event.content.parts:
                response_text = event.content.parts[0].text
                print(f"   {response_text}")
                if event.is_final_response():
                    final_response = response_text
    
    session = await session_service.get_session(
        app_name=APP_NAME,
        user_id=USER_ID, 
        session_id=session_id,
    )
    
    print(f"{'='*60}")
    print(f"📊 工作流完成 - 会话状态 ({session_id}):")
    print(f"{'='*60}")
    for key, value in session.state.items():
        print(f"   {key}: {value}")
    print(f"🎯 最终响应: {final_response}")
    return final_response

测试案例

我们可以准备多个测试案例来验证工作流：

async def main_notebook():
    test_requests = [
        "我需要审批750美元用于团队午餐和庆祝活动",
        "请审批3,000美元用于会议门票和差旅费用",
        "我需要审批12,000美元用于关键软件许可证续订"
    ]
    
    for i, request in enumerate(test_requests, 1):
        current_session_id = f"approval_session_notebook_{456 + i -1}"
        await session_service.create_session(
            app_name=APP_NAME,
            user_id=USER_ID,
            session_id=current_session_id
        )
        print(f"创建会话: {current_session_id}")
        await run_approval_workflow_notebook(request, current_session_id)

try:
    asyncio.run(main_notebook())
    agentops.end_trace(end_state="Success")
except Exception as e:
    print(f"错误: {e}")
    agentops.end_trace(end_state="Error")

总结

本文展示的基于 AgentOps-AI 和 Google ADK 的人工审批工作流具有以下特点：

1. 模块化设计：每个处理阶段由专门的 Agent 负责，职责单一
2. 人工干预：在关键决策点引入人类判断
3. 状态管理：使用会话状态在不同 Agent 间传递数据
4. 可观测性：通过 AgentOps 追踪整个工作流执行过程
5. 灵活性：可以轻松调整审批逻辑或添加新的处理环节

这种架构特别适合需要人工参与的自动化流程，如费用审批、内容审核等业务场景。开发者可以根据实际需求扩展或修改工作流中的各个环节。