尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

领域驱动设计结合AI Agent:用Harness与Loop构建可控智能系统

领域驱动设计结合AI Agent:用Harness与Loop构建可控智能系统 这次我们来看一个关于“Harness”、“Loop”和“领域模型”如何与AI结合的技术思考。这个话题不是介绍一个具体的开源工具而是探讨一种工程实践和思维框架。它源于对传统“敏捷开发”培训的反思并试图通过“领域模型”这一核心概念来构建更可控、更高效的AI应用开发流程。如果你正在为AI项目的需求飘忽、效果不稳定、难以融入现有业务系统而头疼这篇文章或许能提供一个新的视角。简单来说这讨论的是如何将软件工程中成熟的“领域驱动设计”DDD思想与当前火热的AI Agent、Human-in-the-Loop等模式相结合打造一个结构清晰、边界明确、且人类专家能深度参与的AI系统开发“缰绳”Harness和“循环”Loop。其核心价值不在于提供一个即插即用的代码库而在于提供一套方法论帮助团队降低AI的“幻觉”风险提升复杂AI任务的可控性与可交付性。本文将带你梳理几个关键概念什么是Harness Engineering和Loop Engineering它们如何与领域模型结合这种思路对实际开发AI应用如基于LangGraph构建Agent工作流有什么具体指导意义我们会从理念剖析到实践映射探讨如何将这些思想落地从而更稳健地驾驭AI能力。1. 核心能力速览理念框架而非具体工具首先需要明确本文讨论的“Harness”和“Loop”并非某个特定的DeepSeek Harness桌面端工具尽管网络热词中有相关搜索而是一种工程范式。下表概括了这种思路的核心要点能力项说明与解读核心理念用领域模型作为“锚点”通过Harness缰绳控制AI行为边界通过Loop循环引入人类反馈与迭代构建可信赖的AI系统。目标问题解决AI应用开发中的需求不明确、输出不可控幻觉、难以调试、无法持续演进等问题。关键组件1.领域模型形式化、结构化的业务知识表示是AI与业务对话的“通用语言”。2.Harness一系列约束、验证规则、防护栏确保AI输出符合领域规范。3.LoopHuman-in-the-loop人机回环机制在关键决策点引入人工干预与校正。技术关联与LangGraph、AI Agent工作流设计高度相关。领域模型可定义Agent的状态和工具Harness是状态转移的约束条件Loop是特定节点的人工审核或修正步骤。输出物不是可直接运行的软件而是设计规范、架构图、Prompt模板、验证规则集、工作流定义等。适合场景1. 企业级复杂业务流程的AI辅助或自动化。2. 对准确性、合规性要求高的场景如法律、金融、医疗咨询。3. 需要持续学习和改进的AI系统。不适合场景追求快速原型、一次性简单任务、或对输出容错率极高的娱乐性应用。2. 适用场景与使用边界这种基于领域模型、Harness和Loop的思路主要服务于希望将AI深度集成到复杂业务中的团队。它最适合谁企业架构师与技术负责人需要为AI项目制定可落地、可维护的技术架构。资深后端或AI工程师负责实现关键业务逻辑与AI能力的结合需要降低集成风险。产品经理与业务专家拥有深厚的领域知识需要一种高效的方式将其“灌输”给AI系统并保持对最终输出的控制力。能解决什么问题需求沟通漏斗业务人员用自然语言描述需求开发人员将其翻译为领域模型AI基于领域模型理解任务。领域模型成为三者对齐的中间层。输出质量控制通过Harness如格式校验、逻辑规则、事实核查对AI的原始输出进行过滤和修正确保结果符合业务规则。系统可演进性当业务规则变化时主要修改的是领域模型和Harness规则而非重写大量Prompt或代码使系统更易于维护。风险可控在Loop中设置人工审核节点对于高风险操作如审批、支付、重要内容发布保留最终决定权。需要警惕的边界不是银弹这套方法会引入前期的设计和建模成本不适合“五分钟出一个演示”的场景。它追求的是长期稳定性和可控性而非极致开发速度。依赖领域专家构建高质量的领域模型需要业务专家的深度参与。如果无法获得清晰的领域知识模型本身就会有缺陷。性能开销额外的验证步骤Harness和人工干预Loop会带来延迟和成本。需要在自动化程度和可控性之间取得平衡。伦理与合规当系统用于做重大决策时必须明确Human-in-the-loop的责任归属。Harness规则的设计也需避免引入歧视或偏见。3. 从“敏捷培训”到“AI驾驭”思维转变传统的“敏捷开发”培训强调快速迭代、响应变化但在面对AI项目时我们常常发现“用户故事”难以精准定义AI的行为“冲刺”可能产出的是无法使用的幻觉输出。其根源在于自然语言描述的需求对于AI来说粒度太粗、歧义太多。领域模型在此扮演了“精准需求说明书”的角色。它不是一个模糊的想法而是用类图、状态机、实体关系等形式化或半形式化手段描绘的业务核心。例如一个“保险理赔”领域模型会明确定义“保单”、“报案人”、“损失项”、“审核阶段”、“赔付金额”等实体及其关系、状态和规则。当AI接收到任务时我们不再仅仅给它一段自然语言描述而是同时提供或让其参考这个结构化的领域模型。这极大地缩小了AI的“想象”空间使其输出更有可能落在业务预期的范围内。Harness则是执行层面的保障它像一组单元测试对AI的产出进行断言生成的理赔报告是否包含了所有必需的实体金额计算是否符合业务规则如果不符合则触发修正或进入Loop——请求人类专家介入。这个“建模-约束-反馈”的循环构成了驾驭AI的新“敏捷”实践迭代的不是模糊的功能而是越来越精确的领域模型和越来越健壮的Harness规则。4. 核心概念深度拆解4.1 领域模型AI与业务世界的“对齐层”领域模型是这一切的基石。它不仅仅是数据库表设计更是业务逻辑的载体。形式可以是UML图、JSON Schema、Protobuf定义、甚至是一组精心设计的Python数据类。作用结构化Prompt将领域模型作为上下文提供给大模型引导其按结构思考。例如“请根据以下‘客户’实体结构生成描述{“name”: str, “level”: [‘VIP’, ‘普通’], “orderHistory”: List[Order]}”。输出约束要求AI的输出必须符合某个特定的JSON Schema这本身就是一种最基础的Harness。工具Function/Tool定义在AI Agent框架中领域实体和操作可以转化为Agent可调用的工具。例如“创建理赔单(claim_id, applicant_info, loss_details)”工具。构建建议与业务专家协作从核心子领域开始逐步细化。优先保证核心概念的准确性和一致性。4.2 Harness Engineering为AI套上“缰绳”Harness指的是一整套用于约束、验证和引导AI行为的工程化设施。静态Harness事前约束系统Prompt定义角色、边界和禁忌。例如“你是一个保险理赔助手只能处理车险理赔不回答健康险问题。”输出格式强制要求以JSON、XML或特定Markdown表格格式输出。提供知识库与上下文通过RAG检索增强生成提供准确的参考信息限制AI自由发挥。动态Harness事后验证与修正规则引擎校验对AI输出的结构化数据运行业务规则校验。例如校验理赔金额是否在保单限额内。事实核查调用外部API或查询数据库验证AI输出中的关键事实如日期、编号、条款。逻辑一致性检查检查输出内容内部是否自相矛盾。备用策略当AI输出无法通过Harness时触发降级方案如返回固定提示、转接人工、或使用更保守的模板生成。4.3 Loop Engineering不可或缺的“人类在环”无论Harness多完善对于关键业务人类监督仍是安全网。Loop机制设计了何时、如何引入人工干预。设计模式审核节点在Agent工作流如LangGraph中设置特定节点AI的输出在此节点暂停等待人工批准或修改后才能进入下一步。置信度阈值AI为输出附上一个置信度分数低于阈值时自动转入人工处理队列。异常捕获当Harness中的规则校验失败时自动创建人工工单。主动学习人工纠正的结果被记录下来用于微调模型或优化Harness规则形成闭环。工具支持可以集成内部工单系统、IM工具如钉钉/飞书机器人、或构建专门的人工审核后台。5. 实践映射以LangGraph构建AI Agent工作流为例理论需要落地。我们以当前流行的AI Agent编排框架LangGraph为例看如何将上述思想融入一个具体的“智能客服升级理赔”Agent设计中。场景用户向AI客服描述一起车祸Agent需要自动创建理赔案并初步评估损失。5.1 步骤一定义领域模型状态结构在LangGraph中State是所有节点共享的内存。我们首先用Pydantic模型定义清晰的状态结构这就是我们的核心领域模型。from typing import TypedDict, List, Optional, Literal from pydantic import BaseModel # 定义领域实体 class Policy(BaseModel): policy_id: str type: Literal[car, property] coverage_limit: float class Applicant(BaseModel): name: str contact: str policy_id: str class LossItem(BaseModel): description: str estimated_cost: float category: Literal[vehicle_damage, personal_injury, third_party_property] class Claim(BaseModel): claim_id: str applicant: Applicant policy: Optional[Policy] None loss_items: List[LossItem] [] total_estimated_loss: float 0.0 status: Literal[draft, verified, human_review, approved, rejected] draft # LangGraph的State继承自TypedDict但我们可以嵌入Pydantic模型 class AgentState(TypedDict): user_input: str # 原始用户输入 current_claim: Claim # 核心领域对象 validation_errors: List[str] # Harness校验错误 need_human_review: bool # Loop触发标志 review_reason: str # 需要人工审核的原因5.2 步骤二设计工作流节点与Harness每个节点执行特定功能并可能包含Harness逻辑。from langgraph.graph import StateGraph, END import asyncio # 节点1信息提取与结构化 async def extract_claim_info(state: AgentState): # 利用LLM结合系统Prompt和Claim的Pydantic模型定义从user_input中提取信息 system_prompt 你是一个保险理赔信息提取专家。请从用户描述中提取信息并严格按照提供的JSON格式输出。 只提取你确信的信息不确定的字段留空。 # 这里模拟调用LLM实际使用ChatModel.invoke llm_response await call_llm(system_prompt, state[user_input], output_schemaClaim.schema()) draft_claim Claim.parse_raw(llm_response) state[current_claim] draft_claim return state # 节点2数据验证与丰富Harness核心 async def validate_and_enrich(state: AgentState): claim state[current_claim] errors [] # Harness 1: 关键字段非空校验 if not claim.applicant.policy_id: errors.append(保单号不能为空) if not claim.loss_items: errors.append(至少需要一项损失描述) # Harness 2: 业务规则校验模拟 if claim.policy and claim.policy.type ! car: errors.append(本流程仅处理车险理赔) for item in claim.loss_items: if item.estimated_cost 0: errors.append(f损失项{item.description}的成本估算需大于0) # Harness 3: 外部数据核对模拟查询数据库 if claim.applicant.policy_id: # 模拟根据保单号查询保单详情 fetched_policy await fetch_policy_from_db(claim.applicant.policy_id) if fetched_policy: claim.policy fetched_policy # 校验申请人是否与保单匹配伪代码 if not validate_applicant_match(claim.applicant, fetched_policy): errors.append(申请人信息与保单记录不匹配) else: errors.append(未找到有效保单信息) state[validation_errors] errors state[current_claim] claim # 更新状态 return state # 节点3决策路由是否进入Loop async def decision_routing(state: AgentState): # 根据Harness校验结果决定下一步 if state[validation_errors]: # 如果有错误标记需要人工审核 state[need_human_review] True state[review_reason] f数据校验失败{; .join(state[validation_errors])} elif state[current_claim].total_estimated_loss 10000: # 假设阈值是1万元 # 如果损失金额过大标记需要人工审核 state[need_human_review] True state[review_reason] 预估损失金额超过自动处理阈值 else: state[need_human_review] False return state # 节点4自动处理节点 async def auto_process_claim(state: AgentState): if state[need_human_review]: return state # 如果需要审核直接跳过此节点 # 执行自动创建理赔案、发送确认通知等逻辑 print(f自动处理理赔案 {state[current_claim].claim_id}) state[current_claim].status approved return state # 节点5人工审核节点Loop的入口 async def human_review_node(state: AgentState): if not state[need_human_review]: return state # 在实际系统中这里会将state[current_claim]和state[review_reason]推送至人工审核队列 # 并阻塞或等待回调。此处模拟挂起。 print(f[人工审核待处理] 理赔案 {state[current_claim].claim_id}。原因{state[review_reason]}) # 模拟人工审核后更新状态为批准或拒绝 # state[current_claim].status approved # 或 rejected return state5.3 步骤三组装工作流图将节点连接起来形成可控的工作流。# 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(“extract”, extract_claim_info) workflow.add_node(“validate”, validate_and_enrich) workflow.add_node(“decide”, decision_routing) workflow.add_node(“auto_process”, auto_process_claim) workflow.add_node(“human_review”, human_review_node) # 定义边 workflow.set_entry_point(“extract”) workflow.add_edge(“extract”, “validate”) workflow.add_edge(“validate”, “decide”) # 条件边根据need_human_review决定路由 workflow.add_conditional_edges( “decide”, lambda state: “human_review” if state[“need_human_review”] else “auto_process”, {“human_review”: “human_review”, “auto_process”: “auto_process”} ) workflow.add_edge(“auto_process”, END) workflow.add_edge(“human_review”, END) # 人工审核后流程结束或可设计重新进入自动流程 # 编译图 app workflow.compile()5.4 步骤四运行与测试# 初始化状态 initial_state: AgentState { “user_input”: “我的车昨天追尾了前保险杠撞坏了我的保单号是IC20240001。我感觉修车要花七八千吧。”, “current_claim”: None, “validation_errors”: [], “need_human_review”: False, “review_reason”: “” } # 运行工作流 final_state app.invoke(initial_state) print(“最终理赔案状态”, final_state[“current_claim”].status) print(“是否需要人工审核”, final_state[“need_human_review”])通过这个例子可以看到领域模型Claim,Policy等Pydantic类定义了数据的形状Harness体现在validate_and_enrich节点的各种校验规则中Loop机制通过decision_routing和human_review_node实现。整个流程结构清晰可控性强。6. 工程化建议与最佳实践将这种模式投入生产环境还需要考虑以下几点领域模型的版本化与管理业务规则会变领域模型也需要版本化。考虑将核心的Pydantic模型或JSON Schema定义在独立的版本化包中。Harness规则的外部化配置不要将业务规则硬编码在节点函数里。可以将规则存储在数据库或配置文件中以便业务人员能在不重启服务的情况下进行调整。Loop的人工交互设计为人工审核节点设计友好的后台界面清晰地展示AI的输入、输出、置信度、以及触发审核的原因并提供便捷的修改和批准工具。可观测性与调试在工作流每个节点记录详细的日志包括输入、输出、耗时、触发的规则等。这对于排查AI的“诡异”行为至关重要。测试策略单元测试针对每个Harness规则函数编写测试。集成测试模拟完整工作流使用历史案例或构造的边界案例进行端到端测试。冒烟测试每天用一组固定输入跑一遍核心流程监控输出是否发生漂移。安全与合规输入过滤对user_input进行必要的清洗防止提示词注入。输出净化在最终返回结果前对内容进行二次过滤如去除个人隐私信息。审计追踪记录每一次Loop中的人工操作满足合规审计要求。7. 常见问题与排查思路在实施过程中你可能会遇到以下典型问题问题现象可能原因排查方式解决方案AI输出无法解析为领域模型1. Prompt未明确要求格式。2. 领域模型定义太复杂AI不理解。3. AI产生幻觉编造了不存在的字段。1. 检查系统Prompt和Few-shot示例。2. 简化模型先测试最基本字段。3. 查看AI的原始输出。1. 强化输出格式指令使用JSON Schema引导。2. 分步提取先提取简单实体再组合。3. 增加Harness对无法解析的响应进行重试或降级。Harness规则频繁触发导致大量人工工单1. 规则过于严格。2. AI在特定场景下能力不足。3. 业务本身存在大量模糊地带。1. 分析工单数据看哪些规则触发最多。2. 检查触发规则的输入样本。1. 调整规则阈值区分“警告”和“阻断”。2. 针对高频场景优化Prompt或提供更详细的上下文。3. 将模糊规则本身作为人工审核点。工作流执行缓慢1. 单个LLM调用耗时过长。2. 串行节点过多。3. 外部API如数据库查询延迟高。1. 使用异步调用。2. 分析各节点耗时。3. 检查网络和依赖服务状态。1. 设置合理的LLM调用超时。2. 对于无依赖的节点考虑并行执行。3. 为外部调用增加缓存、重试和降级机制。人工审核后如何重新注入流程工作流设计为审核后即结束未考虑回流。检查工作流图的设计。修改图结构让人工审核节点可以修改state后重新路由到auto_process或其他节点。在LangGraph中这可以通过将human_review节点连接到decide节点来实现。8. 总结驾驭AI从“敏捷”到“精密”回顾从“敏捷开发培训”到“Harness与Loop工程”的思考其核心脉络是从追求“快速响应变化”演进到追求“在变化中保持控制”。AI的不确定性是新的挑战而软件工程中已有的抽象、建模、校验和反馈机制正是应对这一挑战的宝贵资产。最值得尝试的起点不是从头构建庞大系统而是在你现有的、最头疼的AI应用场景中挑选一个核心环节尝试为其定义一个简单的领域模型哪怕只是一个JSON Schema并添加一两条关键的Harness校验规则。感受一下这种“先定义轨道再发车”带来的可控性提升。最容易踩的坑试图一次性构建完美、庞大的领域模型。应该采用迭代方式从一个小而准的核心模型开始随着业务理解和AI能力的磨合逐步扩展和修正。下一步方向探索将这种模式与低代码平台结合让业务专家能通过可视化方式定义领域实体和业务规则Harness并自动生成部分Agent工作流代码。这或许是实现AI应用民主化开发和高效迭代的关键。通过领域模型把控AI本质上是将人类的结构化知识转化为机器可执行、AI可理解的约束框架。这不仅是技术架构的升级更是团队协作方式的进化。它要求业务、开发和AI更加紧密地围绕“共识模型”工作而这或许是这个AI时代里一种更高阶的“敏捷”。
返回列表