最近在搭建智能体时，遇到了一个棘手的问题：AI助手在处理复杂业务场景时，经常会做出一些"聪明但错误"的决策。比如在处理退款申请时，AI会根据用户描述直接批准高额退款，但实际上这需要人工审核。这个问题让我深入思考了企业级Agent中Human-in-the-Loop（HITL）的重要性。经过几个月的实践，我总结了一套从单机到分布式的HITL实现方案，希望能帮助大家避免踩坑。

一、为什么企业级Agent必须要HITL？

1、企业场景的刚性需求

在企业环境中，AI的错误往往意味着真金白银的损失。我们在实际部署中发现了几个典型问题：

• 智能客服的幻觉风险：AI可能会承诺无法兑现的服务，或者给出错误的政策解释

• OA流程的合规要求：财务审批、人事变动等流程必须有人工最终确认

• 高危工具的管控需求：数据库操作、API调用等需要严格的权限控制

2、技术挑战

实现HITL看似简单，但在企业级场景下面临诸多技术挑战：

• 流程中断恢复：如何在人工审核后无缝恢复AI流程

• 状态持久化：长时间的审核流程需要可靠的状态存储

• 多会话管理：支持多用户并发的审核流程

• 故障容错：系统重启后如何恢复中断的会话

二、单机模式：基于LangGraph的核心实现

让我们先从单机模式开始，理解HITL的核心机制。

1、核心机制详解

LangGraph提供了三个关键机制来实现HITL：

from langgraph.graph import StateGraph, Commandfrom langgraph.prebuilt import interruptfrom langgraph.checkpoint.postgres import PostgresCheckpointimport asyncioclass AgentState:    def __init__(self):        self.user_query = ""        self.llm_output = ""        self.human_feedback = ""        self.final_result = ""def llm_processing_node(state: AgentState):    """LLM处理节点"""    # 模拟LLM处理    state.llm_output = f"针对查询'{state.user_query}'的AI回答：建议批准退款1000元"    return statedef human_review_node(state: AgentState):    """人工审核节点 - 关键的中断点"""    review_data = {        "question": "请审核以下AI建议是否合理：",        "ai_suggestion": state.llm_output,        "user_query": state.user_query,        "timestamp": "2025-07-12 14:30:00"    }        # 这里是关键：interrupt会挂起流程并返回审核数据    decision = interrupt(review_data)        # 根据人工反馈决定下一步    if decision and decision.get("action") == "approve":        state.human_feedback = decision.get("feedback", "已批准")        return Command(goto="approved_node")    else:        state.human_feedback = decision.get("feedback", "已拒绝")        return Command(goto="rejected_node")def approved_node(state: AgentState):    """批准后的处理"""    state.final_result = f"审核通过：{state.llm_output}\n人工反馈：{state.human_feedback}"    return statedef rejected_node(state: AgentState):    """拒绝后的处理"""    state.final_result = f"审核拒绝，需要重新处理\n人工反馈：{state.human_feedback}"    return state# 构建图def create_agent_graph():    workflow = StateGraph(AgentState)        # 添加节点    workflow.add_node("llm_processing", llm_processing_node)    workflow.add_node("human_review", human_review_node)    workflow.add_node("approved", approved_node)    workflow.add_node("rejected", rejected_node)        # 定义流程    workflow.add_edge("llm_processing", "human_review")    workflow.add_edge("approved", "__end__")    workflow.add_edge("rejected", "__end__")        workflow.set_entry_point("llm_processing")        return workflow# 使用PostgreSQL进行状态持久化def setup_checkpoint():    return PostgresCheckpoint(        connection_string="postgresql://user:pass@localhost/hitl_db",        table_name="agent_checkpoints"    )# 客户端使用示例async def run_agent_with_hitl():    # 设置检查点    checkpoint = setup_checkpoint()        # 编译图    graph = create_agent_graph().compile(checkpointer=checkpoint)        # 创建会话配置    config = {"configurable": {"thread_id": "user_session_001"}}        # 初始状态    initial_state = AgentState()    initial_state.user_query = "我的订单有问题，需要退款"        try:        # 启动流程        result = await graph.ainvoke(initial_state, config)        print("流程完成，结果：", result.final_result)            except Exception as e:        if "interrupt" in str(e):            print("流程已中断，等待人工审核...")                        # 获取中断数据            state_snapshot = graph.get_state(config)            interrupt_data = state_snapshot.next[0].interrupt                        print("审核数据：", interrupt_data)                        # 模拟人工审核            human_decision = {                "action": "approve",  # 或 "reject"                "feedback": "退款金额合理，同意批准"            }                        # 恢复流程            final_result = await graph.ainvoke(human_decision, config)            print("审核完成，最终结果：", final_result.final_result)if __name__ == "__main__":    asyncio.run(run_agent_with_hitl())

2、关键注意事项

在实际使用中，我踩过几个坑，分享给大家：

• 中断恢复位置：流程会从中断节点（human_review_node）恢复，而不是从调用interrupt的位置恢复

• 状态修改时机：不要在interrupt调用之前修改状态，否则可能导致数据不一致

• 线程ID管理：thread_id是状态持久化的关键，必须保证唯一性和可追踪性

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】

​​​​​​​

三、工具调用的HITL管控模式

在企业环境中，工具调用往往是风险最高的环节。我们总结了两种主要的管控模式：

1、集中看守模式

这种模式适合有严格安全审计要求的场景：​​​​​​​

from typing import List, Dict, Anyfrom langgraph.prebuilt import interrupt# 高风险工具配置HIGH_RISK_TOOLS = [    "database_write",    "file_delete",     "api_payment",    "user_privilege_change"]def tool_guardian_node(state: AgentState):    """工具看守节点 - 统一审批高风险工具"""    if not hasattr(state, 'pending_tool_calls'):        return state        # 检查是否有高风险工具调用    risky_calls = []    safe_calls = []        for tool_call in state.pending_tool_calls:        if tool_call['tool_name'] in HIGH_RISK_TOOLS:            risky_calls.append(tool_call)        else:            safe_calls.append(tool_call)        # 直接执行安全工具    for safe_call in safe_calls:        result = execute_tool(safe_call)        state.tool_results.append(result)        # 高风险工具需要审批    if risky_calls:        review_data = {            "message": "检测到高风险工具调用，需要审批：",            "risky_tools": risky_calls,            "context": state.user_query,            "risk_level": "HIGH"        }                approval = interrupt(review_data)                if approval and approval.get("action") == "approve":            # 执行被批准的工具            approved_tools = approval.get("approved_tools", [])            for tool_call in approved_tools:                result = execute_tool(tool_call)                state.tool_results.append(result)        else:            # 记录拒绝信息            state.tool_results.append({                "status": "rejected",                "message": "高风险工具调用被拒绝",                "feedback": approval.get("feedback", "")            })        return statedef execute_tool(tool_call: Dict[str, Any]) -> Dict[str, Any]:    """模拟工具执行"""    tool_name = tool_call['tool_name']    args = tool_call.get('args', {})        # 这里实现具体的工具逻辑    if tool_name == "database_write":        return {"status": "success", "message": f"数据库写入完成: {args}"}    elif tool_name == "api_payment":        return {"status": "success", "message": f"支付接口调用完成: {args}"}        return {"status": "unknown", "message": f"未知工具: {tool_name}"}

2、自我管理模式

这种模式适合工具开发团队有自治能力的场景：​​​​​​​

from functools import wrapsfrom typing import Callable, Anydef human_in_the_loop(risk_level: str = "medium", auto_approve_conditions: List[str] = None):    """HITL装饰器 - 为工具添加人工审核能力"""    def decorator(func: Callable) -> Callable:        @wraps(func)        def wrapper(*args, **kwargs):            # 构建审核数据            review_data = {                "tool_name": func.__name__,                "args": args,                "kwargs": kwargs,                "risk_level": risk_level,                "description": func.__doc__ or "无描述"            }                        # 检查自动批准条件            if auto_approve_conditions:                for condition in auto_approve_conditions:                    if eval(condition, {"args": args, "kwargs": kwargs}):                        return func(*args, **kwargs)                        # 人工审核            response = interrupt(review_data)                        if response and response.get("action") == "approve":                # 可能修改参数                modified_args = response.get("modified_args", args)                modified_kwargs = response.get("modified_kwargs", kwargs)                return func(*modified_args, **modified_kwargs)            else:                return {                    "status": "rejected",                    "message": "工具调用被拒绝",                    "feedback": response.get("feedback", "")                }                return wrapper    return decorator# 使用示例@human_in_the_loop(    risk_level="high",    auto_approve_conditions=["kwargs.get('amount', 0) < 100"]  # 小额自动批准)def tavily_search(query: str, max_results: int = 5):    """使用Tavily进行网络搜索"""    # 模拟搜索API调用    import requests        response = requests.post(        "https://api.tavily.com/search",        json={            "api_key": "your-api-key",            "query": query,            "max_results": max_results        }    )        return response.json()@human_in_the_loop(risk_level="critical")def database_delete(table: str, condition: str):    """删除数据库记录 - 高风险操作"""    # 模拟数据库操作    sql = f"DELETE FROM {table} WHERE {condition}"    print(f"执行SQL: {sql}")    return {"status": "success", "affected_rows": 10}

四、分布式系统实现：FastAPI架构

单机模式在小规模应用中表现良好，但面对企业级多用户场景，我们需要分布式架构。

1、架构设计​​​​​​​

from fastapi import FastAPI, HTTPException, BackgroundTasksfrom pydantic import BaseModelfrom typing import Dict, Any, Optionalimport asyncioimport uuidfrom datetime import datetimeapp = FastAPI(title="企业级HITL Agent服务")# 数据模型class AgentRequest(BaseModel):    user_id: str    query: str    context: Optional[Dict[str, Any]] = Noneclass AgentResponse(BaseModel):    session_id: str    status: str  # "running", "interrupted", "completed", "failed"    result: Optional[Any] = None    interrupt_data: Optional[Dict[str, Any]] = None    timestamp: datetimeclass HumanFeedback(BaseModel):    action: str  # "approve", "reject", "modify"    feedback: Optional[str] = None    modified_params: Optional[Dict[str, Any]] = None# 全局会话存储sessions: Dict[str, Dict[str, Any]] = {}class SessionManager:    """会话管理器"""        @staticmethod    def create_session(user_id: str) -> str:        """创建新会话"""        session_id = f"{user_id}_{uuid.uuid4().hex[:8]}"                # 创建独立的Agent实例        graph = create_agent_graph().compile(            checkpointer=setup_checkpoint()        )                sessions[user_id] = {            "agent": graph,            "session_id": session_id,            "status": "created",            "last_response": None,            "config": {"configurable": {"thread_id": session_id}}        }                return session_id        @staticmethod    def get_session(user_id: str) -> Optional[Dict[str, Any]]:        """获取会话"""        return sessions.get(user_id)        @staticmethod    def update_session_status(user_id: str, status: str, response: AgentResponse = None):        """更新会话状态"""        if user_id in sessions:            sessions[user_id]["status"] = status            if response:                sessions[user_id]["last_response"] = response@app.post("/agent/invoke", response_model=AgentResponse)async def invoke_agent(request: AgentRequest, background_tasks: BackgroundTasks):    """启动Agent会话"""    try:        # 创建会话        session_id = SessionManager.create_session(request.user_id)                # 启动异步任务        background_tasks.add_task(            process_agent_task,            request.user_id,            request.query,            request.context or {}        )                return AgentResponse(            session_id=session_id,            status="running",            timestamp=datetime.now()        )            except Exception as e:        raise HTTPException(status_code=500, detail=str(e))async def process_agent_task(user_id: str, query: str, context: Dict[str, Any]):    """异步处理Agent任务"""    session = SessionManager.get_session(user_id)    if not session:        return        try:        # 准备初始状态        initial_state = AgentState()        initial_state.user_query = query                # 执行Agent        result = await session["agent"].ainvoke(initial_state, session["config"])                # 成功完成        response = AgentResponse(            session_id=session["session_id"],            status="completed",            result=result.final_result,            timestamp=datetime.now()        )                SessionManager.update_session_status(user_id, "completed", response)            except Exception as e:        if "interrupt" in str(e):            # 流程中断，等待人工审核            state_snapshot = session["agent"].get_state(session["config"])            interrupt_data = state_snapshot.next[0].interrupt if state_snapshot.next else None                        response = AgentResponse(                session_id=session["session_id"],                status="interrupted",                interrupt_data=interrupt_data,                timestamp=datetime.now()            )                        SessionManager.update_session_status(user_id, "interrupted", response)        else:            # 其他错误            response = AgentResponse(                session_id=session["session_id"],                status="failed",                result=f"错误: {str(e)}",                timestamp=datetime.now()            )                        SessionManager.update_session_status(user_id, "failed", response)@app.post("/agent/resume", response_model=AgentResponse)async def resume_agent(user_id: str, feedback: HumanFeedback):    """恢复中断的Agent流程"""    session = SessionManager.get_session(user_id)    if not session:        raise HTTPException(status_code=404, detail="会话不存在")        if session["status"] != "interrupted":        raise HTTPException(status_code=400, detail="会话状态不是中断状态")        try:        # 注入人工反馈        feedback_data = {            "action": feedback.action,            "feedback": feedback.feedback,            "modified_params": feedback.modified_params        }                # 恢复流程        result = await session["agent"].ainvoke(feedback_data, session["config"])                response = AgentResponse(            session_id=session["session_id"],            status="completed",            result=result.final_result,            timestamp=datetime.now()        )                SessionManager.update_session_status(user_id, "completed", response)        return response            except Exception as e:        raise HTTPException(status_code=500, detail=str(e))@app.get("/agent/status/{user_id}", response_model=AgentResponse)async def get_agent_status(user_id: str):    """获取Agent状态"""    session = SessionManager.get_session(user_id)    if not session:        raise HTTPException(status_code=404, detail="会话不存在")        last_response = session.get("last_response")    if not last_response:        return AgentResponse(            session_id=session["session_id"],            status=session["status"],            timestamp=datetime.now()        )        return last_response# 健康检查@app.get("/health")async def health_check():    return {"status": "healthy", "active_sessions": len(sessions)}if __name__ == "__main__":    import uvicorn    uvicorn.run(app, host="0.0.0.0", port=8000)

2、客户端使用示例​​​​​​​

import requestsimport timeimport jsonclass HITLClient:    """HITL客户端"""        def __init__(self, base_url: str = "http://localhost:8000"):        self.base_url = base_url        self.session_id = None        def start_agent(self, user_id: str, query: str, context: dict = None):        """启动Agent"""        response = requests.post(            f"{self.base_url}/agent/invoke",            json={                "user_id": user_id,                "query": query,                "context": context or {}            }        )                if response.status_code == 200:            data = response.json()            self.session_id = data["session_id"]            return data        else:            raise Exception(f"启动失败: {response.text}")        def check_status(self, user_id: str):        """检查状态"""        response = requests.get(f"{self.base_url}/agent/status/{user_id}")                if response.status_code == 200:            return response.json()        else:            raise Exception(f"状态检查失败: {response.text}")        def submit_feedback(self, user_id: str, action: str, feedback: str = None):        """提交人工反馈"""        response = requests.post(            f"{self.base_url}/agent/resume",            params={"user_id": user_id},            json={                "action": action,                "feedback": feedback            }        )                if response.status_code == 200:            return response.json()        else:            raise Exception(f"反馈提交失败: {response.text}")        def run_interactive_session(self, user_id: str, query: str):        """运行交互式会话"""        print(f"启动Agent会话: {query}")                # 启动Agent        result = self.start_agent(user_id, query)        print(f"会话ID: {result['session_id']}")                # 轮询状态        while True:            status = self.check_status(user_id)            print(f"当前状态: {status['status']}")                        if status['status'] == 'completed':                print(f"任务完成: {status['result']}")                break            elif status['status'] == 'interrupted':                print("需要人工审核:")                print(json.dumps(status['interrupt_data'], indent=2, ensure_ascii=False))                                # 模拟人工审核                action = input("请输入审核结果 (approve/reject): ").strip()                feedback = input("请输入反馈信息: ").strip()                                # 提交反馈                result = self.submit_feedback(user_id, action, feedback)                print(f"反馈已提交，继续执行...")            elif status['status'] == 'failed':                print(f"任务失败: {status['result']}")                break                        time.sleep(1)# 使用示例if __name__ == "__main__":    client = HITLClient()        # 运行交互式会话    client.run_interactive_session(        user_id="user_001",        query="我需要查询数据库并删除过期记录"    )

五、故障恢复策略

企业级系统必须考虑各种故障场景，我们的恢复策略包括：

客户端故障恢复​​​​​​​

import redisimport jsonfrom typing import Dict, Anyclass ResilientSessionManager:    """具备故障恢复能力的会话管理器"""        def __init__(self, redis_url: str = "redis://localhost:6379"):        self.redis_client = redis.from_url(redis_url)        self.sessions = {}        def persist_session_metadata(self, user_id: str, session_data: Dict[str, Any]):        """持久化会话元数据"""        key = f"session:{user_id}"        metadata = {            "session_id": session_data["session_id"],            "status": session_data["status"],            "timestamp": session_data.get("timestamp", ""),            "last_response": session_data.get("last_response")        }                self.redis_client.setex(            key,            86400,  # 24小时过期            json.dumps(metadata, default=str)        )        def recover_session(self, user_id: str) -> Optional[Dict[str, Any]]:        """恢复会话"""        key = f"session:{user_id}"        metadata = self.redis_client.get(key)                if not metadata:            return None                metadata = json.loads(metadata)                # 重建Agent实例        graph = create_agent_graph().compile(            checkpointer=setup_checkpoint()        )                # 从checkpoint恢复状态        config = {"configurable": {"thread_id": metadata["session_id"]}}                session_data = {            "agent": graph,            "session_id": metadata["session_id"],            "status": metadata["status"],            "config": config,            "last_response": metadata.get("last_response")        }                self.sessions[user_id] = session_data        return session_data        def get_or_recover_session(self, user_id: str) -> Optional[Dict[str, Any]]:        """获取或恢复会话"""        # 先尝试从内存获取        session = self.sessions.get(user_id)        if session:            return session                # 尝试从Redis恢复        return self.recover_session(user_id)# 在FastAPI中使用resilient_session_manager = ResilientSessionManager()@app.post("/agent/invoke", response_model=AgentResponse)async def invoke_agent_resilient(request: AgentRequest, background_tasks: BackgroundTasks):    """具备故障恢复能力的Agent启动"""    try:        # 检查是否有可恢复的会话        existing_session = resilient_session_manager.get_or_recover_session(request.user_id)                if existing_session and existing_session["status"] == "interrupted":            # 返回中断状态，等待人工审核            return AgentResponse(                session_id=existing_session["session_id"],                status="interrupted",                interrupt_data=existing_session.get("last_response", {}).get("interrupt_data"),                timestamp=datetime.now()            )                # 创建新会话        session_id = resilient_session_manager.create_session(request.user_id)                # 启动异步任务        background_tasks.add_task(            process_agent_task_resilient,            request.user_id,            request.query,            request.context or {}        )                return AgentResponse(            session_id=session_id,            status="running",            timestamp=datetime.now()        )            except Exception as e:        raise HTTPException(status_code=500, detail=str(e))async def process_agent_task_resilient(user_id: str, query: str, context: Dict[str, Any]):    """具备故障恢复能力的Agent任务处理"""    session = resilient_session_manager.get_or_recover_session(user_id)    if not session:        return        try:        # 处理逻辑与之前相同        initial_state = AgentState()        initial_state.user_query = query                result = await session["agent"].ainvoke(initial_state, session["config"])                response = AgentResponse(            session_id=session["session_id"],            status="completed",            result=result.final_result,            timestamp=datetime.now()        )                # 更新会话状态并持久化        session["status"] = "completed"        session["last_response"] = response        resilient_session_manager.persist_session_metadata(user_id, session)            except Exception as e:        if "interrupt" in str(e):            # 获取中断数据            state_snapshot = session["agent"].get_state(session["config"])            interrupt_data = state_snapshot.next[0].interrupt if state_snapshot.next else None                        response = AgentResponse(                session_id=session["session_id"],                status="interrupted",                interrupt_data=interrupt_data,                timestamp=datetime.now()            )                        # 更新会话状态并持久化            session["status"] = "interrupted"            session["last_response"] = response            resilient_session_manager.persist_session_metadata(user_id, session)

六、企业级落地价值与最佳实践

1、模式选型建议

根据我们的实践经验，不同场景下的推荐模式：

场景	推荐模式	原因
金融、医疗等高风险行业	集中看守模式	统一风控，审计友好
研发工具、内部系统	自我管理模式	开发效率高，灵活性强
大型企业混合场景	混合模式	核心业务集中管控，辅助功能自我管理

2、性能优化要点

• 异步处理：使用FastAPI的后台任务避免阻塞

• 状态压缩：定期清理过期的checkpoint数据

• 会话池化：复用Agent实例降低内存开销

• 缓存策略：对频繁访问的会话状态进行缓存

 如何学习AI大模型？

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】

 

第一阶段： 从大模型系统设计入手，讲解大模型的主要方法；

第二阶段： 在通过大模型提示词工程从Prompts角度入手更好发挥模型的作用；

第三阶段： 大模型平台应用开发借助阿里云PAI平台构建电商领域虚拟试衣系统；

第四阶段： 大模型知识库应用开发以LangChain框架为例，构建物流行业咨询智能问答系统；

第五阶段： 大模型微调开发借助以大健康、新零售、新媒体领域构建适合当前领域大模型；

第六阶段： 以SD多模态大模型为主，搭建了文生图小程序案例；

第七阶段： 以大模型平台应用与开发为主，通过星火大模型，文心大模型等成熟大模型构建大模型行业应用。

👉学会后的收获：👈
• 基于大模型全栈工程实现（前端、后端、产品经理、设计、数据分析等），通过这门课可获得不同能力；

• 能够利用大模型解决相关实际项目需求： 大数据时代，越来越多的企业和机构需要处理海量数据，利用大模型技术可以更好地处理这些数据，提高数据分析和决策的准确性。因此，掌握大模型应用开发技能，可以让程序员更好地应对实际项目需求；

• 基于大模型和企业数据AI应用开发，实现大模型理论、掌握GPU算力、硬件、LangChain开发框架和项目实战技能， 学会Fine-tuning垂直训练大模型（数据准备、数据蒸馏、大模型部署）一站式掌握；

• 能够完成时下热门大模型垂直领域模型训练能力，提高程序员的编码能力： 大模型应用开发需要掌握机器学习算法、深度学习框架等技术，这些技术的掌握可以提高程序员的编码能力和分析能力，让程序员更加熟练地编写高质量的代码。

1.AI大模型学习路线图
2.100套AI大模型商业化落地方案
3.100集大模型视频教程
4.200本大模型PDF书籍
5.LLM面试题合集
6.AI产品经理资源合集

👉获取方式：
😝有需要的小伙伴，可以保存图片到wx扫描二v码免费领取【保证100%免费】🆓

​