企业官网建设流程全解析

1. 四种模式整体设计：从固定流水线到自主决策

先说个背景。最近一两年，身边聊Agent的人越来越多了，但真正能把Agent落到生产环境、稳定跑上几个月的团队其实并不多。我见过太多团队一上来就搞“让大模型自己决定一切”的自主Agent，结果线上问题一大堆，最后灰溜溜退回普通代码流程。也有团队从头到尾只用固定Workflow，明明可以自动化的判断非要写死if-else，白白浪费了大模型的理解能力。

这两种极端其实指向同一个问题：没有把“流程编排”这件事想清楚。以我实际做过的几个项目来看，Agent和Workflow并不是“二选一”的关系，而是不同场景下的不同工具。把它们按模式拆开，再按场景组合，才是工程上最稳的做法。

行业内逐渐收敛出一个共识：自动化系统按控制方式可以分成四种典型模式。

这四种模式不是谁替代谁的关系。一个成熟的生产系统往往会混合使用：外层用Workflow做骨架，中间插入路由节点做判断，关键环节用状态机管理流转，局部交给自主Agent去探索。下面我按模式拆开讲，每个模式都会配上实践代码和踩坑记录。

2. 核心选型逻辑：为什么这个场景要选那种模式

很多初学者拿到需求就直接开写Prompt，让大模型输出JSON然后执行。这种“裸奔式”做法在小Demo里看着挺酷，一到生产环境就露馅：输出格式不稳定、关键字段丢失、循环调用停不下来。选型逻辑第一条就是：能用代码保证的确定性，就不要交给模型。

2.1 什么时候选固定流程Workflow

固定流程适合“流程明确、步骤固定、不需要智能判断”的场景。典型如CI/CD流水线：代码检查、单元测试、构建镜像、部署环境，这些步骤的顺序是死的，不需要模型参与。再比如定时数据同步、批量文件处理、报表生成，这些都是标准管道。

实现方式可以直接用Python代码写一个Pipeline类，或者用现成的编排框架。关键知识点是：

• 每个步骤是一个独立函数，输入输出固定
• 步骤之间通过数据结构传递上下文
• 失败重试、超时控制放在框架层统一处理

2.2 什么时候选路由模式

路由模式解决的核心问题只有一个：这个请求该走哪条处理链路。比如客服工单进来，先判断是“退款问题”“技术故障”还是“产品咨询”，再分发给不同的处理流程。如果所有工单都塞给同一个Prompt处理，模型能力会被稀释，成本也会飙升。

路由的实现有两种主流方式：

• Embedding相似度路由：把用户输入向量化，和预定义的类别中心算余弦相似度，取最高分。适合类别少、边界清晰的场景。
• LLM语义路由：让模型直接输出类别标签，适合类别多、边界模糊、需要推理的场景。

实际项目中我倾向于Embedding做粗筛、LLM做兜底裁决，两者配合。路由决策只调一次模型，之后的处理链路全是确定性的代码，这个“只调一次”非常关键，控制了成本也控制了风险。

2.3 什么时候选状态机模式

状态机适合业务流程有“状态流转”和“人工参与”的场景。最典型的就是工单系统和审批流：工单有“待受理”“处理中”“待用户确认”“已关闭”这些状态，状态之间通过事件触发迁移，而且随时可能插入人工操作。这时候如果还是线性流水线，根本没法表达“用户回复后回到处理中”这种循环。

状态机模式的落地要点是状态持久化和事件驱动。每条工单的当前状态必须存数据库，收到用户消息、超时提醒、管理员操作等事件后，根据当前状态决定下一步。

2.4 什么时候才轮到自主Agent

自主Agent是上限最高但下限也最低的模式。它适合任务目标明确但路径未知的场景：让模型自己拆解步骤、自己调用工具、自己判断结果。典型场景包括竞品信息搜集、复杂数据分析、多源资料整理等。

算法的核心循环是：思考 -> 调用工具 -> 观察结果 -> 再思考。每一步模型看到工具返回结果后更新自己的计划。落地时必须做三重防护：

1. 工具调用次数上限（比如单任务最多10次）
2. 操作白名单（模型只能调用预先注册的API，不能任意执行系统命令）
3. 最终结果校验（关键结果必须有结构化校验逻辑）

3. 实操过程与核心环节实现：一个工单系统的四模式演进

为了让你更直观地理解四种模式怎么落地，我设计了一个贯穿案例：某公司的客服工单自动分拣与处理系统。一开始需求很简单，工单进来后自动分类、自动回复常见问题、复杂问题转人工。但随着需求深入，这个系统会依次经历四种模式，每种模式解决前一阶段的痛点。

3.1 第一阶段：固定流程实现自动分级

最简单的做法是把处理流程固定成三步：接收工单 -> 提取关键词 -> 按规则入库。先用代码实现一个基础的Pipeline。

from dataclasses import dataclass, field from typing import Callable, Any import time @dataclass class WorkflowContext: """流程上下文，所有步骤共享的数据容器""" input_data: dict results: dict = field(default_factory=dict) errors: list = field(default_factory=list) class Pipeline: """一个极简的固定流程编排器""" def __init__(self): self._steps: list[tuple[str, Callable]] = [] def add_step(self, name: str, fn: Callable): self._steps.append((name, fn)) return self def run(self, input_data: dict) -> WorkflowContext: ctx = WorkflowContext(input_data=input_data) for name, fn in self._steps: try: print(f"[Pipeline] 执行步骤: {name}") ctx.results[name] = fn(ctx) except Exception as e: ctx.errors.append((name, str(e))) # 关键设计：步骤失败是否中断？ # 生产环境建议做成可配置，默认中断并抛出 raise return ctx def step_extract_keywords(ctx: WorkflowContext) -> dict: """从工单标题中提取关键词""" title = ctx.input_data.get("title", "") # 粗糙但够用：按空格切分，剔除停用词 stopwords = {"我", "的", "是", "了", "请", "帮忙", "一下"} keywords = [w for w in title.split() if w not in stopwords] return {"keywords": keywords} def step_analyze_level(ctx: WorkflowContext) -> dict: """根据关键词判断紧急程度""" keywords = ctx.results["step_extract_keywords"]["keywords"] urgent_flags = {"故障", "宕机", "损失", "紧急"} level = "P1" if any(k in urgent_flags for k in keywords) else "P2" return {"level": level, "assigned_to": "二线专家" if level == "P1" else "一线客服"} # 组装 pipe = Pipeline() pipe.add_step("step_extract_keywords", step_extract_keywords) pipe.add_step("step_analyze_level", step_analyze_level) result = pipe.run({"title": "生产环境故障 紧急 请处理"}) print(result.results)

这一段代码的核心设计点有两个：一是用WorkflowContext统一传递数据，步骤之间不直接耦合；二是允许配置失败行为。固定流程是四种模式里唯一不需要模型参与的，所以它可以作为所有上层模式的基础设施。

固定流程的问题也很明显：只能按写好的规则走。遇到“我的银行卡昨天还能用今天突然付不了款了”这种句子，关键词提取完全失效。这时候就要引入路由模式。

3.2 第二阶段：路由模式自动分类分派

路由模式的核心是让大模型做一次单点决策，把工单分类，然后交给不同分支处理。这一步的效果直接决定了后续所有环节的质量。

from openai import OpenAI client = OpenAI() # 假设已经配置好API密钥 SYSTEM_PROMPT = """ 你是工单分类专家。请把用户的工单分到以下类别之一： - refund: 退款/退货/发票类问题 - technical: 技术故障/系统报错/无法使用 - consultation: 产品咨询/价格询问/功能疑问 - complaint: 投诉/不满/要求升级处理 只输出类别名称，不要输出其他任何内容。 """ def route_ticket(title: str, content: str) -> str: """路由决策：模型只负责输出一个分类标签""" resp = client.chat.completions.create( model="gpt-4o-mini", temperature=0, # 路由决策必须确定性优先 messages=[ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": f"标题：{title}\n内容：{content}"} ] ) category = resp.choices[0].message.content.strip() # 兜底：模型输出不合法类别时，走默认分类 valid = {"refund", "technical", "consultation", "complaint"} return category if category in valid else "consultation"

路由模式有四个容易踩的坑，值得展开说。

第一个坑是temperature必须设成0。分类任务不需要创造性，一旦模型“发挥”出新的类别名，后续所有分支都会失效。我在实践中见过模型输出“refunds”带个s的情况，就是因为疏忽。

第二个坑是必须加白名单校验。模型的输出不可控性比想象中严重，它可能输出“这是一个关于退款的问题”这种完整句子。校验逻辑不是可选项而是必需项，不能匹配就直接重试或给默认值。

第三个坑是类别不能太多。实际操作中我发现，类别超过10个时基于LLM的路由准确率明显下降。如果确实要分很多类，应该先做一级分类再做二级分类，层级控制在两到三层之内。

第四个坑是缓存策略。同一个客户反复提交相似工单，每次都调模型做分类非常浪费。最好按工单标题做语义去重，短时间内重复出现直接命中历史分类。

路由模式解决的是“智能判断”的问题，但它是无状态的。工单处理到一半需要人工复核，或者需要等客户补充信息，线性路由就撑不住了。于是需要引入状态机。

3.3 第三阶段：状态机模式让流程可暂停可恢复

工单场景里，用户提交工单后可能中途补充信息，客服处理时可能临时挂起，处理完可能被用户驳回。这些“非直线”的流程就是状态机的典型应用场景。我用transitions这个轻量库来实现。

from transitions import Machine import json class Ticket: """工单对象：状态由状态机管理，数据保存在对象中""" states = ["submitted", "classifying", "processing", "waiting_user", "resolved", "closed", "escalated"] def __init__(self, ticket_id: str, title: str, content: str): self.ticket_id = ticket_id self.title = title self.content = content self.category = None self.notes = [] self.machine = Machine( model=self, states=Ticket.states, initial="submitted", transitions=[ # 提交后进入分类 {"trigger": "start_classify", "source": "submitted", "dest": "classifying"}, # 分类完成进入处理 {"trigger": "finish_classify", "source": "classifying", "dest": "processing"}, # 处理中需要用户补充信息 {"trigger": "ask_user", "source": "processing", "dest": "waiting_user"}, # 用户补充后恢复处理 {"trigger": "receive_user_reply", "source": "waiting_user", "dest": "processing"}, # 处理完成 {"trigger": "resolve", "source": "processing", "dest": "resolved"}, # 用户确认后关闭 {"trigger": "close", "source": "resolved", "dest": "closed"}, # 任意状态都可以升级 {"trigger": "escalate", "source": ["processing", "waiting_user"], "dest": "escalated"}, ], auto_transitions=False, send_event=True, queued=True, # 同一时间只处理一个变迁，防止并发冲突 ) def add_note(self, note: str): self.notes.append(note) # 测试状态流转 tk = Ticket("T001", "数据库连接超时", "业务系统频繁报错") tk.start_classify() tk.finish_classify() tk.ask_user() print("当前状态:", tk.state) # waiting_user tk.receive_user_reply() tk.resolve() tk.close() print("最终状态:", tk.state) # closed

状态机落地的关键点不只是“状态定义”，而是以下三个容易被忽略的工程细节。

第一个是并发锁。同一个工单可能同时被用户消息和客服操作触发状态变迁。queued=True能保证事件原子化，但多实例部署时还得靠数据库的行锁来实现跨进程的并发控制。具体做法是更新状态前先执行UPDATE tickets SET version=version+1 WHERE id=? AND version=?，乐观锁防止旧状态覆盖新状态。

第二个是状态持久化。状态不能只存在内存对象里，必须同步写入数据库。同时要记录全量状态迁移日志，这个日志是之后排障和责任追溯的依据。实际应用中这个日志通常是审计表，字段包含工单ID、旧状态、新状态、操作人、触发事件、时间戳。

第三个是超时自动迁移。这是状态机比路由模式强大很多的地方。比如waiting_user状态超过48小时没有收到客户回复，可以设计一个定时任务触发auto_close事件，自动迁移到closed状态。这种“时间维度”的控制是普通代码很难优雅实现的。

工单系统到这里已经能稳定服务常规场景了。但它依然解决不了“用户描述不清楚但问题真的很复杂”的情况。比如一个用户说“系统用不了”，类别可以分到technical，但具体是什么故障、该怎么排查，需要模型主动去查知识库、查日志、尝试多种方案。这就进入了自主Agent的领域。

3.4 第四阶段：自主Agent模式处理开放式问题

自主Agent在工单场景里的定位不是替代全套流程，而是在“处理中”状态里嵌入一个子Agent，让它自主收集信息、尝试方案、输出处理建议，最后由系统根据建议决定下一步。这样设计的好处是风险可控——Agent就算胡言乱语，也不会直接破坏工单状态。

from typing import Any TOOLS = { "search_kb": "搜索知识库，输入关键词，返回相关文档列表", "get_ticket_detail": "查看工单的完整详情，输入工单号，返回结构化数据", "check_service_status": "检查核心服务的健康状态，输入服务名，返回正常运行或异常", "escalate_to_human": "把工单标记为需要人工介入，输入原因说明", } class AssistantAgent: """一个极简的自主Agent：思考-行动-观察循环""" def __init__(self, model="gpt-4o", max_steps=8): self.model = model self.max_steps = max_steps self.memory = [] # 保存对话历史 def run(self, task: str) -> dict: messages = [ {"role": "system", "content": self._system_prompt()}, {"role": "user", "content": task} ] for i in range(self.max_steps): resp = self._call_llm(messages) reply = resp.choices[0].message.content action = self._parse_action(reply) if action["type"] == "final": return {"status": "done", "answer": action["result"], "steps": i + 1} if action["type"] == "tool_call": result = self._execute_tool(action) messages.append({"role": "assistant", "content": reply}) messages.append({"role": "tool", "content": json.dumps(result, ensure_ascii=False)}) else: return {"status": "error", "answer": "无法理解模型输出", "steps": i + 1} return {"status": "max_steps_exceeded", "answer": "超过最大执行步数", "steps": self.max_steps} def _system_prompt(self) -> str: tool_desc = "\n".join(f"- {name}: {desc}" for name, desc in TOOLS.items()) return f""" 你是一个技术售后助理。你可以使用以下工具来帮助用户解决问题： {tool_desc} 请按以下格式输出： - 如果需要调用工具：调用 工具名(参数) - 如果已经得到最终答案：最终答案 [内容] 注意：每次只调用一个工具，等待结果后再决定下一步。 """ def _parse_action(self, reply: str) -> dict: # 真实系统中建议用结构化输出，这里为演示用简单匹配 if reply.startswith("调用 "): parts = reply[3:].split("(", 1) tool_name = parts[0].strip() arg = parts[1].rstrip(")") if len(parts) > 1 else "" return {"type": "tool_call", "tool": tool_name, "arg": arg} if reply.startswith("最终答案"): return {"type": "final", "result": reply[4:].strip()} return {"type": "unknown"} def _execute_tool(self, action: dict) -> str: tool = action.get("tool", "") arg = action.get("arg", "") if tool not in TOOLS: return f"错误：未知工具 {tool}" # 这里实际调用具体函数 return f"工具 {tool} 执行成功，参数：{arg}，返回示例结果"

这个实现当然是简化过的，生产级Agent应该注意下面几个细节。

工具注册表用{工具名: 描述}字典的好处是，Prompt里的工具列表可以用字典动态生成，新增工具不用改代码。但工具描述直接决定了模型选工具的成功率，描述写得太泛模型会乱试，写得太具体又显得啰嗦。经验是描述里同时包含“输入什么”和“输出什么”，外加一句典型使用场景。

max_steps=8不是随便设的。太少了模型来不及完成调研，太多了失败时成本不可控。我做过一组实验，8步左右的完成率与成本比是最合适的。另外还有一个隐形机制：每轮调用前检查是否连续调用了同一个工具且参数相同，如果是，判定循环，直接终止。这能拦截大概率是模型陷入重复的情况。

Agent的结果返回给上层状态机后，上层还是走“处理中 -> 需要用户确认 -> 关闭”的既有流程。Agent在里面只是一个高级工具，不是一个失控的自治体。这个分层思路是整个工单系统稳定运行的关键。

4. 常见问题与排查技巧实录

项目从第一阶段的裸Pipeline演进到第四阶段的分层系统，前后经过三轮重构，踩了不少坑。这一节挑几个最典型的写下来，给后来者提个醒。

4.1 模型输出格式不稳定的处理方案

不管用哪种模式，大模型的输出都不可能做到100%按格式。实践中我采用的方案是三层兜底：

• 第一层：Prompt里给输出示例，给出1到2个“正例+反例”
• 第二层：程序侧做正则或解析逻辑校验，解析失败自动重试一次（温度降为0）
• 第三层：重试仍失败就使用默认值，并在日志里标记一个解析异常计数

这三层下来，格式错误率从最初实测的8%左右降到了0.5%以下。很多团队只做了第一层，然后靠人工盯日志，效率很低。

4.2 状态机与数据库不同步的解法

状态机跑在应用内存里，数据库存的是持久化记录，两者可能因为异常退出而出现不一致。排障方法如下：

排查时先检查审计表，看最后一次迁移日志记录的是什么。如果发现状态停留在processing但数据库写的是waiting_user，多半是ask_user事件触发了但后续的数据库更新事务回滚了。修复方案是写一个一致性问题巡检脚本，定期扫描工单表与审计表的状态差，把不一致的记录重新对齐到一个合法状态。

防患于未然的做法是状态迁移和数据库更新放进同一个事务：先更新数据库，成功后再触发状态机事件。反过来如果先触发事件再写库，内存状态和数据库的先后关系就会有间隙。

4.3 Agent死循环与成本失控

自主Agent模式最常见的问题就是跑起来停不下来。我在项目里遇到过模型反复调用同一个搜索工具长达3小时的情况，最后是消耗金额触发了告警才被发现。

我的应对策略是给Agent套上三重保险丝：

• 总步数限制：硬上限8步，到点强制收尾
• 单调工具调用检测：统计最近5次动作，如果同一个工具加同一个参数出现两次以上，直接终止
• 每步超时：单次工具执行超过30秒就返回超时错误，连续超时3次自动结束

这三条同时开，实测下来死循环基本被拦截在萌芽阶段。另外，生产环境强烈建议把Agent每次调用的明细日志打全（时间、模型、输入、输出、工具名、参数、耗时、费用)，这不仅仅是排障需要，也是之后做成本归因的依据。

4.4 四种模式的常见问题速查

这四类问题对应了四种模式各自的薄弱点。理解了它们，你在设计阶段就能提前打补丁，而不是上线后疲于救火。

5. 可靠性保障与成本工程实践

模式选完、代码写完，真实的考验才刚开始。生产环境最关心的永远是可靠性和成本，这两个问题决定了项目能不能长期跑下去。

5.1 单一模型依赖的风险与对策

所有模式都在调用大模型，而大模型本身有不确定性。我在实践中逐步形成了模型无关化的策略：在代码与模型之间抽象出一个模型网关，把所有调用封装成统一接口。这样做有两个直接好处——上游模型出问题可以快速切换到备用模型，另外一个模型的成本策略可以单独调整而不影响业务代码。

网关还会统一做重试、超时、熔断。重试指数退避从1秒开始，最多重试3次。熔断逻辑是连续5次失败就熔断30秒，不再发请求，给下游模型喘息时间。

5.2 成本估算与控制

四种模式的成本差异巨大，估算时需要分别建模。以我项目里的数据作为参考：

成本控制的实战技巧有几个。首先是把高成本模式用在刀刃上，Agent只处理那些路由判断为复杂的工单，简单工单直接走固定流程。其次是给Prompt瘦身，不要为了“让它更智能”塞大量上下文，实测下来多数任务500字以内就能说清楚。最后是设置日消耗上限和单工单成本上限告警，超出自动降级到人工处理。

5.3 四模式混合编排的推荐架构

最后给出一个我验证过的混合架构作为参考，兼顾稳定性与智能化：

入口 -> 路由模式(模型分类) ├─ 退款/咨询类 -> 固定流程Workflow（模板回复+自动操作） ├─ 技术故障类 -> 进入State Machine │ └─ 处理中 -> 自主Agent（调研问题+尝试方案） │ └─ Agent结果返回后回到状态机继续流转 └─ 投诉升级类 -> 直接转人工并通知主管

这套架构的外层是一个固定流程框架，确保每条工单必有归属；路由节点做一次高价值判断；状态机管好生命周期，让人工能在任意环节介入；Agent只在有限步数内、有限的工具集里发挥自主性。四层各司其职，每层都有兜底，稳定性与智能化可以兼得。

我试过把路由和Agent合并、也试过完全自治的Agent方案，最终在这套混合架构下，工单处理的自动化率跑到了70%以上，剩余30%人工介入本身就是业务要求——金融级客户确实需要人来确认。这就是真实世界里“自动化”的边界，不是100%模型全干，而是该自动的自动，该人工的人工，各得其所。