2026 年 10 月 24 日至 25 日，2026 长沙 1024 程序员日暨 AI 构建者大会在湖南湘江新区长沙智谷举行。大会以“具身同行，智构未来”为主题，由 CSDN 主办，长沙软件园有限公司协办，线下参会人数超过 1500 人。从现场讨论看，AI 正在从“辅助写代码”的工具，进一步走向“可调度、可编排、可交付”的生产力基础设施。蒋涛在开幕式上提出“硅基时间”概念：当 AI 承担大量执行工作后，团队扩展方式正在从单纯增加人力转向“加 Token”，开发者的核心价值也从基础编码转向目标定义、质量判断、智能体调度与工作流设计。
在此背景下，“确定性工作流”成为 AI 工程化落地的关键能力。所谓确定性工作流，并不是要求 AI 模型完全消除随机性，而是通过流程编排、规则校验、人工审核、异常恢复和版本控制，让 AI 能力进入真实业务后能够稳定、可追踪、可审计地产生结果。飞猪技术专家梁宇航在分享中也指出，AI 能力接入履约系统时，需要区分“服务调用成功”“结果被采用”和“业务完成”三个层次，不能只关注模型是否返回了内容，更要关注结果能否被现有系统承接。
一、为什么 AI 应用需要确定性工作流
大语言模型擅长理解、生成和推理，但直接将其接入生产系统时，常会遇到三类问题。
第一，输出不稳定。模型可能生成格式错误、字段缺失或语义偏差的内容。例如在审核、审批、报表生成等场景中，输出必须满足明确结构，不能只靠“尽量写对”。
第二，流程不可追踪。若 AI 调用、规则判断、人工复核、系统写入都散落在不同脚本中，一旦出现问题，很难定位是模型错误、数据错误还是流程错误。
第三，异常难以恢复。真实业务中，网络超时、接口失败、数据缺失、权限不足都很常见。如果没有重试、回滚、检查点和人工兜底，AI 流程很容易在复杂环境中失效。
因此，确定性工作流的目标不是替代人的判断，而是把人的经验、业务规则和系统接口组织成一条可靠的执行链。AI 负责处理非结构化信息、生成候选方案、提取结构化数据；工作流负责保证步骤完整、状态清晰、异常可控；人负责在关键节点进行审核和决策。
二、确定性工作流的 Python 实现思路
下面以一个“AI 会议纪要生成与审核工作流”为例，展示如何用 Python 构建一个轻量但结构清晰的确定性工作流。该示例不依赖复杂框架，重点体现节点化、状态化、可校验和可恢复的设计思想。
importjsonimporttimefromdatetimeimportdatetimefromtypingimportAny,Dict,List,Optionalfromdataclassesimportdataclass,field,asdictfromenumimportEnumclassWorkflowStatus(str,Enum):PENDING="pending"RUNNING="running"WAITING_REVIEW="waiting_review"COMPLETED="completed"FAILED="failed"@dataclassclassWorkflowContext:"""工作流运行上下文，所有节点只读写约定字段，避免隐式状态扩散。"""run_id:strstatus:WorkflowStatus=WorkflowStatus.PENDING raw_text:str=""extracted:Dict[str,Any]=field(default_factory=dict)validated:bool=Falsereview_comment:Optional[str]=Noneapproved:Optional[bool]=Noneoutput:Optional[Dict[str,Any]]=Nonelogs:List[str]=field(default_factory=list)error:Optional[str]=Nonecreated_at:str=field(default_factory=lambda:datetime.now().isoformat())deflog(self,message:str):self.logs.append(f"[{datetime.now().isoformat()}]{message}")classBaseNode:"""工作流节点基类。"""def__init__(self,name:str):self.name=namedefrun(self,ctx:WorkflowContext)->WorkflowContext:raiseNotImplementedErrorclassNormalizeInputNode(BaseNode):"""输入规范化节点：清洗空值、限制长度、记录元信息。"""defrun(self,ctx:WorkflowContext)->WorkflowContext:ctx.status=WorkflowStatus.RUNNING ctx.log(f"节点开始:{self.name}")text=(ctx.raw_textor"").strip()ifnottext:ctx.status=WorkflowStatus.FAILED ctx.error="输入文本为空"ctx.log("输入校验失败：文本为空")returnctxiflen(text)>5000:text=text[:5000]ctx.log("输入文本已截断至 5000 字符")ctx.raw_text=text ctx.extracted["input_length"]=len(text)ctx.log("输入规范化完成")returnctxclassAIExtractNode(BaseNode):"""AI 提取节点：模拟从非结构化文本中提取结构化信息。"""defrun(self,ctx:WorkflowContext)->WorkflowContext:ifctx.status==WorkflowStatus.FAILED:returnctx ctx.log(f"节点开始:{self.name}")# 实际工程中这里会调用 LLM API，例如生成会议主题、待办、风险点等。# 为了保持示例可运行，这里使用规则模拟结构化输出。text=ctx.raw_text.lower()topics=[]todos=[]if"上线"intextor"发布"intext:topics.append("产品上线计划")if"测试"intext:topics.append("测试安排")if"bug"intextor"缺陷"intext:topics.append("缺陷修复")if"跟进"intextor"负责"intext:todos.append("确认责任人并跟进进度")if"明天"intextor"下周"intext:todos.append("补充明确时间节点")ctx.extracted.update({"topics":topicsor["常规事项"],"todos":todosor["待补充"],"confidence":0.85,})ctx.log("AI 提取完成")returnctxclassValidateOutputNode(BaseNode):"""校验节点：检查 AI 输出是否满足确定性要求。"""REQUIRED_KEYS=["topics","todos","confidence"]defrun(self,ctx:WorkflowContext)->WorkflowContext:ifctx.status==WorkflowStatus.FAILED:returnctx ctx.log(f"节点开始:{self.name}")extracted=ctx.extracted missing=[kforkinself.REQUIRED_KEYSifknotinextracted]ifmissing:ctx.status=WorkflowStatus.FAILED ctx.error=f"缺少必要字段:{missing}"ctx.log("校验失败")returnctxifnotisinstance(extracted.get("topics"),list)ornotextracted["topics"]:ctx.status=WorkflowStatus.FAILED ctx.error="topics 不能为空列表"returnctxifnotisinstance(extracted.get("confidence"),(int,float)):ctx.status=WorkflowStatus.FAILED ctx.error="confidence 类型不正确"returnctx ctx.validated=Truectx.log("输出校验通过")returnctxclassHumanReviewNode(BaseNode):"""人工审核节点：低置信度或高风险输出进入人工确认。"""defrun(self,ctx:WorkflowContext)->WorkflowContext:ifctx.status==WorkflowStatus.FAILED:returnctx ctx.log(f"节点开始:{self.name}")confidence=ctx.extracted.get("confidence",0)ifconfidence>=0.9:ctx.approved=Truectx.review_comment="高置信度自动通过"ctx.log("自动审核通过")else:ctx.status=WorkflowStatus.WAITING_REVIEW ctx.log("进入人工审核队列")returnctxdefapprove(self,ctx:WorkflowContext,approved:bool,comment:str="")->WorkflowContext:ifctx.status!=WorkflowStatus.WAITING_REVIEW:ctx.log("当前状态不需要人工审核")returnctx ctx.approved=approved ctx.review_comment=commentor("人工通过"ifapprovedelse"人工驳回")ctx.log(f"人工审核结果: approved={approved}")returnctxclassPublishNode(BaseNode):"""发布节点：生成最终输出并标记完成。"""defrun(self,ctx:WorkflowContext)->WorkflowContext:ifctx.status==WorkflowStatus.FAILED:returnctx ctx.log(f"节点开始:{self.name}")ifnotctx.validated:ctx.status=WorkflowStatus.FAILED ctx.error="未通过校验，禁止发布"returnctxifctx.approvedisnotTrue:ctx.status=WorkflowStatus.FAILED ctx.error="未获得审核通过"returnctx ctx.output={"run_id":ctx.run_id,"status":"completed","summary":{"topics":ctx.extracted.get("topics"),"todos":ctx.extracted.get("todos"),"confidence":ctx.extracted.get("confidence"),"review_comment":ctx.review_comment,},"finished_at":datetime.now().isoformat(),}ctx.status=WorkflowStatus.COMPLETED ctx.log("工作流执行完成")returnctxclassDeterministicWorkflow:"""确定性工作流引擎：按顺序执行节点，失败即停止，支持恢复。"""def__init__(self,nodes:List[BaseNode]):self.nodes=nodesdefrun(self,ctx:WorkflowContext)->WorkflowContext:ctx.status=WorkflowStatus.RUNNING ctx.log("工作流启动")fornodeinself.nodes:ifctx.statusin(WorkflowStatus.FAILED,WorkflowStatus.WAITING_REVIEW):breaktry:ctx=node.run(ctx)exceptExceptionase:ctx.status=WorkflowStatus.FAILED ctx.error=str(e)ctx.log(f"节点异常:{node.name}, error={e}")breakreturnctxdefresume(self,ctx:WorkflowContext)->WorkflowContext:"""从当前状态继续执行，适合人工审核后恢复。"""ifctx.status==WorkflowStatus.WAITING_REVIEW:returnself.run(ctx)ifctx.statusin(WorkflowStatus.COMPLETED,WorkflowStatus.FAILED):ctx.log("当前状态不可恢复")returnctxreturnself.run(ctx)defto_dict(self,ctx:WorkflowContext)->Dict[str,Any]:returnasdict(ctx)defdemo():workflow=DeterministicWorkflow([NormalizeInputNode("normalize_input"),AIExtractNode("ai_extract"),ValidateOutputNode("validate_output"),HumanReviewNode("human_review"),PublishNode("publish"),])ctx=WorkflowContext(run_id="WF-20261027-001",raw_text="明天下午讨论产品上线计划，测试团队需要跟进缺陷修复，负责人明天确认排期。",)ctx=workflow.run(ctx)# 模拟人工审核ifctx.status==WorkflowStatus.WAITING_REVIEW:ctx=workflow.nodes[3].approve(ctx,approved=True,comment="人工确认通过")ctx=workflow.resume(ctx)print(json.dumps(workflow.to_dict(ctx),ensure_ascii=False,indent=2))if__name__=="__main__":demo()三、代码解析
这段代码的核心不是“调用大模型”，而是把 AI 调用放进一个可控流程里。
WorkflowContext负责承载运行状态。所有节点都通过同一个上下文读写数据，而不是依靠全局变量或隐式参数。这样可以记录status、logs、error、extracted等关键信息，便于问题排查。
NormalizeInputNode体现输入规范化。真实系统中，AI 节点之前通常需要先清洗数据，例如去除空值、限制长度、统一编码、补全必填字段。这样可以减少后续模型和校验节点的异常。
AIExtractNode模拟 AI 能力接入。实际开发中，这里会替换为真实的 LLM 调用，例如提取会议主题、待办事项、风险点、责任人等。为了保持示例可运行，代码中用规则模拟结构化输出。工程上更推荐让模型输出 JSON Schema，并在下一节点进行严格校验。
ValidateOutputNode是确定性工作流的关键。它不信任模型输出，而是检查字段是否存在、类型是否正确、业务规则是否满足。如果校验失败，工作流立即进入失败状态，而不是继续向下游传播错误数据。
HumanReviewNode体现人机协同。置信度较高时可以自动通过，较低时进入人工审核。审核通过后，工作流可以继续执行。这种设计适合审核、审批、合同、财务、政务、运维等高风险场景。
PublishNode负责最终输出。只有在校验通过且审核通过后，才会生成正式结果。这对应真实业务中的“写入数据库”“发送通知”“触发下游系统”等动作。
DeterministicWorkflow提供顺序执行和恢复能力。节点执行失败或等待人工审核时，工作流会暂停，不会继续执行后续节点。resume方法可以在人工审核完成后从当前状态继续执行，从而避免重复运行已完成节点。
四、工程亮点
第一，节点化编排。每个节点只承担单一职责，便于替换、测试和扩展。未来可以把 AIExtractNode 换成不同模型，也可以增加日志节点、监控节点、审计节点。
第二，状态机控制。工作流通过明确状态流转避免“部分成功、部分失败”的混乱情况。失败时立即停止，防止错误数据污染下游系统。
第三，校验前置。AI 输出必须经过结构化校验才能进入发布环节。对于企业场景，这比单纯提高提示词质量更可靠。
第四，人机协同。人工审核不是最后补救，而是工作流中的正式节点。审核意见、置信度、执行日志都可以沉淀下来，用于后续优化模型和规则。
第五，可恢复执行。真实业务中，AI 接口超时、人工审核延迟、外部系统限流都很常见。支持从断点恢复，比重新执行整个流程更节省成本。
五、与大会主题的对应
2026 长沙 1024 程序员日暨 AI 构建者大会强调开发者正在从“写代码”转向“调度智能体”。但智能体要进入真实业务，不能只靠自然语言指令，还需要确定性的执行边界。飞猪机票履约实践也表明，AI 识别、确定性工作流和 Agent 各有适用场景：识别类任务适合接入服务，流程类任务适合沉淀为可运行工作流，复杂探索类任务更适合由 Agent 承担。
对开发者而言，确定性工作流并不是降低 AI 创造力，而是让 AI 创造力能够被业务系统安全接纳。未来，开发者更需要掌握三类能力：一是把业务目标拆成可执行节点；二是为 AI 输出设计校验标准和审核机制；三是把模型、工具、规则、人工操作组织成可维护、可观测、可迭代的系统。
六、后续可扩展方向
在实际项目中，该工作流还可以进一步升级。例如引入 DAG 引擎支持并行节点；增加检查点持久化，使进程重启后仍可恢复；接入真实 LLM API 并支持模型降级；增加 Prometheus 指标采集，监控节点耗时、失败率和人工审核率；同时为每个工作流生成版本快照，便于回滚和审计。这样，Python 脚本就能从演示级原型逐步演进为生产级 AI 工作流服务。
