什么是HitL人机在环大模型执行过程中暂停交给人类输入反馈再继续执行。两种常见 HITL 模式工具调用前请求人类审批是否允许调用工具Agent 思考一半需要人类补充信息 / 修正答案再继续工具调用前请求人类审批from typing import Optional class HumanInTheLoopAgent: def __init__(self): self.task_history [] def llm_simulate(self, prompt: str) - str: 模拟大模型输出 print(f\n [模拟大模型思考] 输入{prompt}) if 分析用户报告 in prompt: return 需要调用数据库查询用户数据需要人类审批是否执行该操作。 elif 拿到人类审批 in prompt: return ✅ 已获得人类许可开始执行查询报告已生成完成。 else: return f大模型回复基于输入[{prompt}]完成处理 def human_input(self, hint: str) - str: HITL 交互暂停等待人类输入 print(\n *60) print(f【人类介入 Human‑in‑the‑Loop】{hint}) user_input input( 请你的输入(人类反馈): ) print(*60) return user_input def run(self, user_query: str): print(f\n Agent启动用户请求{user_query} ) self.task_history.append({role:user,content:user_query}) # Step1第一轮大模型推理 resp1 self.llm_simulate(user_query) print(f Agent输出{resp1}) self.task_history.append({role:agent,content:resp1}) # 判断是否需要人机介入 if 需要人类审批 in resp1: # HITL 触发暂停等待人类 human_feedback self.human_input(模型想要执行数据库查询请审批输入 yes 允许 / 输入其他拒绝也可以写你的修改意见) self.task_history.append({role:human,content:human_feedback}) # Step2把人类反馈丢回大模型继续执行 new_prompt f拿到人类审批结果:{human_feedback}原始任务{user_query} resp2 self.llm_simulate(new_prompt) print(f\n 接收人类反馈后继续执行{resp2}) self.task_history.append({role:agent,content:resp2}) print(\n✅任务全部结束完整对话历史) for item in self.task_history: print(item) if __name__ __main__: agent HumanInTheLoopAgent() # 运行demo agent.run(分析用户报告查询数据库获取用户隐私数据)Agent 思考一半需要人类补充信息 / 修正答案再继续 LangChain Human‑in‑the‑Loop工具调用前人工确认 import os from dotenv import load_dotenv from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langchain.agents import create_agent from langchain.agents.middleware import wrap_tool_call current_file_path os.path.abspath(__file__) script_folder os.path.dirname(os.path.dirname(os.path.dirname(current_file_path))) env_path os.path.join(script_folder, .env.local) load_dotenv(dotenv_pathenv_path) # print(os.getenv(DEEPSEEK_BASE_URL)) # print(env_path) tool def query_database(user_id:str)-str: 查询数据库获取用户敏感数据 return f数据库返回用户[{user_id}]的敏感信息xxx隐私数据 tools [query_database] llm ChatOpenAI(modelos.getenv(DEEPSEEK_MODEL_CHAT), api_keyos.getenv(DEEPSEEK_API_KEY), base_urlos.getenv(DEEPSEEK_BASE_URL)) # 自定义中间件在工具真正执行前拦截触发人工确认HITL wrap_tool_call def human_in_loop(request, handler): tool_call request.tool_call print(f\n⚠️ 模型准备调用工具{tool_call[name]} 参数{tool_call[args]}) human_ok input(是否允许调用该工具 yes/no) if human_ok.strip().lower() ! yes: # 人类拒绝执行该工具调用返回拒绝信息让模型继续推理 return { role: tool, content: 人类拒绝执行该工具调用, tool_call_id: tool_call[id], } # 人类批准继续执行工具 return handler(request) executor create_agent( modelllm, toolstools, system_prompt你是助手按需调用工具, middleware[human_in_loop], ) if __name__ __main__: res executor.invoke({messages: [{role: user, content: 查询id1001用户的数据库信息}]}) print(\n最终结果, res[messages][-1].content)