什么是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_path=env_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(model=os.getenv("DEEPSEEK_MODEL_CHAT"), api_key=os.getenv("DEEPSEEK_API_KEY"), base_url=os.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( model=llm, tools=tools, system_prompt="你是助手，按需调用工具", middleware=[human_in_loop], ) if __name__ == "__main__": res = executor.invoke({"messages": [{"role": "user", "content": "查询id=1001用户的数据库信息"}]}) print("\n最终结果：", res["messages"][-1].content)