智能体与 LLM 应用的低代码平台市场呈现出百花齐放的态势，每个平台都有其独特的定位和优势。选择哪个平台，往往取决于你的核心需求、技术背景以及项目的最终目标。在本章的后续内容中，我们将重点介绍并实操三个各具代表性的平台：Dify、n8n 和 Coze。在此之前，我们先对它们进行一个概要性的介绍。

• Dify 是一个开源的、功能全面的 LLM 应用开发与运营平台，旨在为开发者提供从原型构建到生产部署的一站式解决方案。它融合了后端服务和模型运营的理念，支持 Agent 工作流、RAG Pipeline、数据标注与微调等多种能力。对于追求专业、稳定、可扩展的企业级应用而言，Dify 提供了坚实的基础。
• n8n 本质上是一个开源工作流自动化工具，而非纯粹的 LLM 平台。近年来，它积极集成了 AI 能力。n8n 的强项在于“连接”。它拥有数百个预置的节点，可以轻松地将各类 SaaS 服务、数据库、API 连接成复杂的自动化业务流程。你可以在这个流程中嵌入 LLM 节点，使其成为整个自动化链路中的一环。其通用自动化能力是独一无二的。不过，其学习曲线也相对陡峭。
• 由字节跳动推出的 Coze，主打零代码/低代码的 Agent 的构建体验，让不具备编程背景的用户也能轻松创造。Coze 拥有极其友好的可视化界面，用户可以像搭建乐高积木一样，通过拖拽插件、配置知识库和设定工作流来创建智能体。其内置了极为丰富的插件库，并支持一键发布到抖音、飞书、微信公众号等多个主流平台，极大地简化了分发流程。

智能体与 LLM 应用的低代码平台呈现多元化发展，不同平台各具特色，选择需结合核心需求、技术背景和项目目标：

• 平台	定位与核心优势	适用场景	特点
  Coze	字节跳动推出，主打零代码 / 低代码 Agent 构建	快速原型验证、非技术用户	可视化界面，拖拽式操作，丰富插件库，支持一键发布到抖音、飞书等多平台
  Dify	开源的 LLM 应用开发与运营平台	企业级应用、复杂业务逻辑	支持 Agent 工作流、RAG Pipeline、数据标注与微调，提供从原型到生产的一站式解决方案
  n8n	开源工作流自动化工具，集成 AI 能力	深度业务集成、自动化流程	数百个预置节点，擅长连接各类 SaaS 服务与 API，学习曲线较陡峭

快速原型验证、非技术用户: 优先选择 Coze；企业级应用、复杂业务逻辑: 优先选择 Dify；深度业务集成、自动化流程: 优先选择 n8n。低代码平台并非要取代代码开发，而是提供了一种互补的选择。在实际项目中，我们完全可以根据不同阶段的需求灵活切换:用低代码平台快速验证想法，用代码实现精细化控制;用平台处理标准化流程，用代码处理特殊逻辑。这种"混合开发"的思维，才是智能体工程化的最佳实践。

一个框架的本质，是提供一套经过验证的“规范”。它将所有智能体共有的、重复性的工作（如主循环、状态管理、工具调用、日志记录等）进行抽象和封装，让我们在构建新的智能体时，能够专注于其独特的业务逻辑，而非通用的底层实现。

低代码平台是代码开发的互补选择，实际项目中可采用 “混合开发” 模式 —— 用低代码快速验证想法，用代码实现精细化控制；用平台处理标准化流程，用代码处理特殊逻辑。智能体开发平台的资源是支撑 AI 应用的 “数字基础设施”，包含五大核心类别，共同构成 “感知 - 决策 - 执行 - 反馈” 的闭环系统：

• 资源类型	定义	核心功能	价值定位	关键指标
  知识库 (RAG)	专业领域知识的结构化存储	提供权威信息源，减少模型 “幻觉”	智能体的 “数字大脑”，决定知识深度	内容新鲜度、覆盖率、准确性
  工具接口 / 插件	连接外部系统的 API 通道	扩展能力边界，实现数据获取与操作	智能体的 “感知器官” 和 “执行手臂”	调用成功率、响应延迟、错误率
  提示词 (Prompt)	引导模型行为的指令模板	定义角色、规范输出、控制流程	智能体的 “行为准则”，决定输出质量	清晰度、完整性、引导效果
  工作流 (Workflow)	任务执行的逻辑编排	定义步骤、控制条件、协调多组件	智能体的 “行动蓝图”，决定处理效率	节点效率、分支合理性、异常处理
  记忆系统	存储交互历史与状态	保持上下文连贯性，支持个性化服务	智能体的 “记忆能力”，实现持续交互	上下文相关性、存储效率

在 Coze 平台中，插件、知识库、数据库、提示词库、音色等都属于资源的范畴。插件是一个工具集合，一个插件内可以包含一个或多个工具（API）。Coze 提供了简单易用的知识库功能，支持智能体与用户自己的数据进行交互。用户可以在知识库中存储和管理数据，无论是处理 PDF 中的数十万字还是网站中的实时信息，只需创建一个知识库，机器人就可以访问相关知识。

一个完整的插件包含以下几个部分：

• 元数据：作用是让大语言模型理解每个工具输入或输出参数的含义。因此，在工具的元数据内提供详细的参数说明，可以让大语言模型更准确地使用工具。
• 工具实现：包含具体的 API 调用逻辑和数据处理代码。开发者可以使用 JavaScript 或 Python 等语言编写工具的实现代码。
• 测试配置：用于验证工具的功能是否正常工作。开发者可以设置测试输入参数，并查看工具的输出结果。

大模型由于其工作原理，虽然可以回答通用的问题，但在专业技术的垂直领域中，模型知识的技术深度、更新及时性和准确性非常有限。如果对于模型生成的文本准确性要求较高，例如企业智能客服、高精尖科学技术服务等应用场景，往往需要通过知识库集成私有的知识数据，丰富 AI 应用的知识范围、提高模型回复的可靠性。知识库的核心功能：数据管理与存储；增强检索，全文检索：通过关键词进行内容片段检索和召回，语义检索：基于向量相似度的智能检索，结构化检索：针对表格数据的精确查询。知识增强生成：大语言模型会根据召回的内容片段生成最终的回复内容，确保回答的准确性和相关性。

• 特性	知识库	记忆功能
  数据性质	静态数据	动态数据
  访问权限	开发者管理，可共享	用户私有，不可共享
  使用场景	专业知识、企业数据	用户偏好、会话历史
  更新方式	手动更新	自动学习

提示词（Prompt）是一种自然语言指令，它为大语言模型（LLM）提供任务指导。搭建智能体的第一步就是设置提示词，你可以根据业务需要直接编写提示词，也可以使用提示词模板、引用提示词资源或通过 AI 自动生成提示词。提示词就是给 AI 的 “任务说明书”，它决定了 AI 如何理解用户需求并生成相应的回答。系统提示词：给 AI 定 “人设”，全程影响 AI 的回复风格；用户提示词：具体需求。一个完整的提示词通常包含以下几个核心部分：

• 角色设定（Role），明确 AI 应该扮演的角色和身份，包括：专业背景；技能特长；经验水平；性格特点。
• 任务描述（Task），清晰定义 AI 需要完成的具体任务：任务目标；输入输出要求；质量标准；时间限制。
• 技能说明（Skills），列出 AI 应该具备的能力和可以使用的工具：专业技能；工具使用能力；知识范围；问题解决能力。
• 工作流程（Workflow），说明完成任务的步骤和逻辑：思考过程；执行步骤；决策逻辑；验证方法。
• 输出格式（Format），规定回答的格式和结构：内容组织方式；格式要求；可视化元素；交互方式。
• 约束条件（Constraints），明确 AI 的行为边界和限制：禁止内容，质量要求，伦理准则，合规要求

提示词编写需要清晰明确的指令，结构化的表达方式，提供示例和上下文，迭代优化的思路

• 初始提示：基于初步需求编写提示词
• 测试验证：运行提示词并查看结果
• 分析反馈：评估结果是否符合预期
• 调整优化：修改提示词并再次测试
• 固化模板：将优化后的提示词保存为模板

知识库 (RAG)：构建专业知识基座，知识 vs 记忆区分：知识是开发者维护的共享数据 (如产品手册)，记忆是用户专属的个性化数据 (如浏览历史)。多模态支持：解析 PDF、Word、Excel、图片等 20 + 格式，通过 OCR 提取非结构化内容；混合检索能力：结合语义检索 (理解意图) 与关键词检索 (精确匹配)，提升召回准确率。构建要点：

• 内容规划：按业务领域分类，建立层级结构

• 数据清洗：去重、纠错、实体归一化(如"退款周期"统一为"退货时间")

• 分片处理：合理分段(200-500字/块)，保留上下文关联

• 向量化存储：使用向量数据库+关系型数据库双层存储(语义向量+元数据)

• 检索优化：设计"语义检索+关键词检索+图检索"的混合策略

工具接口与插件：拓展能力边界。设计清晰的接口规范，明确输入输出格式；建立工具调用日志，监控使用频率与成功率；实现 “工具 - 模型 - 知识库” 的闭环协作：先检索→再调用→后整合。 突破 LLM 本身限制，连接企业现有系统 (ERP/CRM/OA)，实现 “语义驱动” 的业务操作，降低系统使用门槛。分类与特点：

• 工具类型	功能定位	典型应用
  数据类工具	获取信息 (搜索、数据库查询、文档解析)	信息检索、市场调研、内容监控
  行动类工具	执行操作 (发送邮件、更新 CRM、调用 API)	流程自动化、订单处理、系统集成
  创意类工具	生成内容 (图像、文案、代码)	营销素材、报告生成、UI 设计
  计算类工具	数据分析与处理 (Pandas、统计)	财务分析、AB 测试、趋势预测

提示词 (Prompt)：塑造智能体行为。核心设计原则：

• 角色清晰：明确身份 (如 “你是一位资深金融分析师”)
• 任务导向：清晰定义目标 (如 “分析用户财务状况并提供投资建议”)
• 约束明确：设定边界 (如 “回答基于知识库内容，不编造信息”)
• 格式规范：指定输出结构 (如 “使用 Markdown 格式，包含标题、要点和结论”)
• **进阶技巧：**使用变量引用："根据用户提供的{query}和知识库中的{product_info}生成回答"，设计多轮对话模板：预设追问策略，引导用户提供完整信息，添加评估指标："确保回答准确率>95%，知识覆盖率>90%"。

工作流 (Workflow)：构建任务执行引擎。可视化编排：通过拖拽节点设计复杂流程，无需编码；条件分支：根据结果动态选择路径 (如 “如果相似度 < 0.7，触发二次搜索”)；多智能体协同：设计智能体间的转交规则 (如客服场景中从基础问答到专家解答)。模块化拆分：将复杂任务分解为单一功能节点，提高复用性；异常处理：设计错误捕获与回退机制，增强系统稳定性；人机协作点：明确何时自动执行、何时需要人工审核。实现跨系统、跨功能的自动化流程，特别适合审批流、数据分析、内容生产等场景，能将处理时间缩短。

智能体的真正威力在于资源间的有机整合，形成 “感知 - 决策 - 执行 - 反馈” 的闭环系统：用户输入 → 意图识别(提示词) → 知识检索(知识库) → 工具调用(接口) → 结果整合 → 输出响应 → 反馈收集(记忆)。

• 知识库提供专业支撑，减少模型臆造，内容新鲜度(更新频率)、覆盖率(关键知识点)、准确性(无矛盾)
• 工具接口扩展能力边界，获取实时数据，调用成功率、响应延迟、错误率、成本效益比
• 提示词引导模型行为，确保输出符合预期，清晰度、完整性、引导效果、输出一致性
• 工作流协调各组件，实现复杂任务，节点效率、分支合理性、异常处理完备性

资源构建的 “黄金三步法”

• 第一步：业务需求分析：识别高频场景：筛选 “高重复、高规则、高价值” 的业务场景；明确目标：如 “提升客服效率 40%” 或 “缩短产品咨询响应时间”；划定边界：确定智能体 “能做什么” 和 “不能做什么”。
• 第二步：资源设计与开发
  • 知识库构建：收集资料 → 清洗整理 → 结构化存储 → 向量化索引 → 检索测试
  • 提示词工程：角色定义 → 任务描述 → 约束设定 → 示例提供 → 迭代优化
  • 工作流设计：流程拆解 → 节点设计 → 条件设置 → 异常处理 → 测试验证
• 第三步：集成与优化，设计 “资源 - 模型 - 用户” 的交互闭环；建立监控指标，持续收集反馈；定期更新资源，保持内容新鲜度

知识库是智能体的 “专业记忆”，其核心是实现 “精准检索 + 知识增强”。从 0 到 1 构建生产级知识库的技术栈通常包括：文档处理→文本分块→向量嵌入→向量数据库→检索优化。

• 文档处理流水线

• # 1. 多格式文档加载（支持PDF/Word/Markdown）
  from langchain.document_loaders import PyPDFLoader, Docx2txtLoader, TextLoader
  from pathlib import Path
  def load_documents(data_dir):
      documents = []
      for file in Path(data_dir).iterdir():
          if file.suffix == ".pdf":
              loader = PyPDFLoader(str(file))
              documents.extend(loader.load())
          elif file.suffix == ".docx":
              loader = Docx2txtLoader(str(file))
              documents.extend(loader.load())
          elif file.suffix in [".txt", ".md"]:
              loader = TextLoader(str(file))
              documents.extend(loader.load())
      return documents
  # 2. 文本清洗（去冗余、规范化）
  import re
  def clean_document(doc):
      # 去除多余空行和特殊字符
      cleaned_text = re.sub(r'\n\s*\n', '\n\n', doc.page_content.strip())
      cleaned_text = re.sub(r'[^\w\s\.\,\?\!]', '', cleaned_text)
      # 保留元数据（来源、页码等关键信息）
      return {
          "text": cleaned_text,
          "metadata": doc.metadata
      }
  # 3. 智能分块（解决上下文断裂问题）
  from langchain.text_splitter import RecursiveCharacterTextSplitter
  def split_documents(cleaned_docs, chunk_size=300, chunk_overlap=50):
      text_splitter = RecursiveCharacterTextSplitter(
          chunk_size=chunk_size,  # 块大小（根据模型上下文窗口调整）
          chunk_overlap=chunk_overlap,  # 重叠部分（保持上下文连贯）
          separators=["\n\n", "\n", ". ", " ", ""]  # 优先按语义分隔符拆分
      )
      chunks = []
      for doc in cleaned_docs:
          splits = text_splitter.split_text(doc["text"])
          # 为每个块添加元数据（溯源用）
          for split in splits:
              chunks.append({
                  "text": split,
                  "metadata": doc["metadata"]
              })
      return chunks
  

• 向量嵌入与存储

• # 1. 向量嵌入（支持本地模型+云服务）
  from sentence_transformers import SentenceTransformer  # 本地开源模型
  # 或使用OpenAI Embeddings（需API密钥）
  # from langchain.embeddings import OpenAIEmbeddings
  
  def embed_texts(texts, model_name="all-MiniLM-L6-v2"):
      model = SentenceTransformer(model_name)
      embeddings = model.encode(texts, show_progress_bar=True)
      return embeddings
  # 2. 向量数据库存储（以FAISS为例，轻量且适合本地部署）
  import faiss
  import numpy as np
  class VectorDB:
      def __init__(self, dimension):
          self.index = faiss.IndexFlatL2(dimension)  # L2距离索引
          self.texts = []  # 存储原始文本，用于检索后返回
          self.metadatas = []  # 存储元数据
      
      def add(self, embeddings, texts, metadatas):
          self.index.add(np.array(embeddings).astype('float32'))
          self.texts.extend(texts)
          self.metadatas.extend(metadatas)
      
      def search(self, query_embedding, top_k=5):
          distances, indices = self.index.search(
              np.array([query_embedding]).astype('float32'), 
              top_k
          )
          results = []
          for i, idx in enumerate(indices[0]):
              results.append({
                  "text": self.texts[idx],
                  "metadata": self.metadatas[idx],
                  "score": 1 - distances[0][i]  # 转换为相似度分数
              })
          return results
  # 3. 构建完整知识库
  def build_knowledge_base(data_dir):
      # 加载文档
      docs = load_documents(data_dir)
      # 清洗
      cleaned_docs = [clean_document(doc) for doc in docs]
      # 分块
      chunks = split_documents(cleaned_docs)
      # 提取文本列表
      texts = [chunk["text"] for chunk in chunks]
      metadatas = [chunk["metadata"] for chunk in chunks]
      # 生成向量
      embeddings = embed_texts(texts)
      # 初始化向量库（维度与模型匹配，all-MiniLM-L6-v2输出384维）
      vdb = VectorDB(dimension=384)
      vdb.add(embeddings, texts, metadatas)
      return vdb
  

• 检索优化技巧，混合检索：结合关键词检索（精确匹配）和语义检索（模糊匹配）

• def hybrid_search(query, vdb, keyword_threshold=0.3):
      # 1. 语义检索
      query_emb = embed_texts([query])[0]
      semantic_results = vdb.search(query_emb, top_k=10)
      # 2. 关键词过滤（从语义结果中筛选含关键词的片段）
      keywords = query.split()
      filtered = []
      for res in semantic_results:
          match_score = sum(1 for kw in keywords if kw.lower() in res["text"].lower())
          if match_score / len(keywords) >= keyword_threshold:
              filtered.append(res)
      return filtered[:5]  # 返回最终Top5结果
  

• 动态分块调整：根据文档类型自动调整块大小（如代码文档用小粒度，小说用大粒度）;元数据过滤：检索时可按来源、日期等元数据筛选（如 “只检索 2023 年后的产品文档”）

工具是智能体的 “四肢”，核心是实现 “函数定义→参数解析→结果处理” 的标准化流程。工程化开发需关注接口稳定性、参数校验和错误处理。

• 自定义工具开发规范

• from pydantic import BaseModel, Field  # 用于参数校验
  from typing import Optional, List
  # 1. 工具参数定义（用Pydantic确保类型安全）
  class WeatherQueryParams(BaseModel):
      city: str = Field(..., description="城市名称，如'北京'")
      date: Optional[str] = Field(None, description="查询日期，格式'YYYY-MM-DD'，默认今天")
  # 2. 工具实现（包含核心逻辑+错误处理）
  class WeatherTool:
      name = "weather_query"
      description = "查询指定城市的天气情况"
      params_schema = WeatherQueryParams  # 关联参数模型
      
      def __call__(self, params: WeatherQueryParams):
          try:
              # 实际场景中调用第三方天气API
              # 此处用模拟数据示例
              import random
              weather_types = ["晴", "多云", "雨", "雪"]
              return {
                  "city": params.city,
                  "date": params.date or "今天",
                  "weather": random.choice(weather_types),
                  "temperature": f"{random.randint(-10, 35)}°C"
              }
          except Exception as e:
              # 标准化错误返回，便于智能体处理
              return {"error": str(e), "code": 500}
  # 3. 工具注册与管理（便于智能体动态调用）
  class ToolRegistry:
      def __init__(self):
          self.tools = {}
      
      def register(self, tool):
          self.tools[tool.name] = tool
          return tool
      
      def get_tool(self, name):
          return self.tools.get(name)
      
      # 生成工具调用说明（供LLM理解如何调用）
      def get_tool_descriptions(self):
          descriptions = []
          for tool in self.tools.values():
              # 提取参数描述
              param_descs = []
              for field in tool.params_schema.__fields__.values():
                  param_descs.append(
                      f"- {field.name}: {field.description}（类型：{field.type_.__name__}）"
                  )
              descriptions.append(f"""
  工具名：{tool.name}
  描述：{tool.description}
  参数：
  {"".join(param_descs)}
  """)
          return "\n".join(descriptions)
  # 注册工具示例
  registry = ToolRegistry()
  registry.register(WeatherTool())
  

• 工具调用流程（与 LLM 集成）

• def call_tool_with_llm(query, llm, registry):
      # 1. 让LLM决定是否调用工具及参数
      prompt = f"""
  你需要处理用户查询：{query}
  可用工具：
  {registry.get_tool_descriptions()}
  如果需要调用工具，请返回JSON格式：{{"tool": "工具名", "params": {{参数键值对}}}}
  如果不需要调用工具，直接返回回答内容。
  """
      llm_response = llm.generate([prompt])  # 假设llm是已初始化的大模型客户端
      response_text = llm_response.generations[0][0].text
      # 2. 解析LLM响应，执行工具调用
      import json
      try:
          tool_call = json.loads(response_text)
          tool = registry.get_tool(tool_call["tool"])
          if tool:
              # 参数校验
              params = tool.params_schema(**tool_call["params"])
              result = tool(params)
              return f"工具调用结果：{result}"
          else:
              return "未找到指定工具"
      except json.JSONDecodeError:
          # 未调用工具，直接返回回答
          return response_text
      except Exception as e:
          return f"工具调用失败：{str(e)}"
  

• 企业级工具开发要点：权限控制：为工具添加访问权限（如 “财务工具仅财务角色可调用”）； 调用限流 ：使用 Redis 实现工具调用频率限制，防止滥用； 日志审计：记录所有工具调用的输入输出，用于问题排查和合规审计。

• def audit_tool_call(tool_name, params, result, user_id):
      import time
      log_entry = {
          "timestamp": time.time(),
          "user_id": user_id,
          "tool_name": tool_name,
          "params": params.dict(),
          "result": result,
          "status": "success" if "error" not in result else "failed"
      }
      # 写入日志系统（如ELK、ClickHouse）
      print(f"Audit log: {log_entry}")  # 示例打印
  

提示词是智能体的 “操作手册”，优秀的提示词应具备结构化、动态化、可维护特性。算法工程师可通过代码生成和优化提示词，避免人工编写的局限性。

• 提示词模板引擎

• from jinja2 import Template  # 强大的模板引擎，支持条件、循环
  
  # 1. 定义智能体角色模板
  agent_template = """
  你是{{ role }}, 负责{{ responsibility }}。
  你的回答需满足以下要求：
  {% for rule in rules %}
  - {{ rule }}
  {% endfor %}
  
  {% if knowledge_base_enabled %}
  你可以使用知识库信息，引用格式为：[来源: {{ doc_source }}]
  {% endif %}
  
  当前上下文：
  {% for msg in context %}
  {{ msg.role }}: {{ msg.content }}
  {% endfor %}
  
  用户当前查询：{{ query }}
  """
  # 2. 动态生成提示词
  def generate_prompt(role, responsibility, rules, query, context=[], knowledge_base_enabled=True):
      template = Template(agent_template)
      return template.render(
          role=role,
          responsibility=responsibility,
          rules=rules,
          query=query,
          context=context,
          knowledge_base_enabled=knowledge_base_enabled
      )
  # 3. 使用示例（电商客服智能体）
  prompt = generate_prompt(
      role="电商客服专员",
      responsibility="解答用户关于商品售后的问题",
      rules=[
          "必须使用知识库中的售后政策回答",
          "回答需包含具体处理流程和时效",
          "语气友好，避免使用专业术语"
      ],
      query="我的订单超过7天了还能退货吗？",
      context=[{"role": "user", "content": "我上周买的衣服不合身"}]
  )
  

• 提示词优化算法，**少样本自动扩充 **：当提示词效果不佳时，自动从历史对话中提取优质示例

• def auto_enrich_prompt(base_prompt, history, top_n=3):
      # 从历史对话中筛选高评分示例
      high_quality_examples = sorted(
          [h for h in history if h.get("rating", 0) >= 4],
          key=lambda x: x["rating"],
          reverse=True
      )[:top_n]
      # 生成示例片段
      examples = "\n".join([
          f"用户：{e['query']}\n回答：{e['response']}"
          for e in high_quality_examples
      ])
      # 插入到基础提示词中
      return f"{base_prompt}\n\n参考示例：\n{examples}"
  

• **动态长度控制 **：根据模型上下文窗口自动调整提示词长度（截断或精简）

• def truncate_prompt(prompt, max_tokens=2048, tokenizer=None):
      if not tokenizer:
          from transformers import GPT2Tokenizer
          tokenizer = GPT2Tokenizer.from_pretrained("gpt2")
      tokens = tokenizer.encode(prompt)
      if len(tokens) <= max_tokens:
          return prompt
      # 截断并保留首尾关键信息
      truncated_tokens = tokens[:int(max_tokens*0.7)] + tokens[-int(max_tokens*0.3):]
      return tokenizer.decode(truncated_tokens)
  

工作流是智能体的 “决策中枢”，核心是实现 “条件分支→并行执行→结果聚合” 的逻辑控制。工程化实现可基于状态机或可视化流程引擎。

• 状态机工作流实现

• from transitions import Machine  # 状态机库，简化状态流转
  class AgentWorkflow:
      states = ["init", "retrieve_knowledge", "call_tool", "generate_response", "end"]
      def __init__(self, agent):
          self.agent = agent  # 智能体实例
          self.machine = Machine(model=self, states=AgentWorkflow.states, initial="init")
          # 定义状态转换规则
          self.machine.add_transition(
              trigger="start", 
              source="init", 
              dest="retrieve_knowledge",
              before="do_retrieve_knowledge"  # 转换前执行的动作
          )
          self.machine.add_transition(
              trigger="needs_tool", 
              source="retrieve_knowledge", 
              dest="call_tool",
              before="do_call_tool"
          )
          self.machine.add_transition(
              trigger="tool_done", 
              source="call_tool", 
              dest="generate_response",
              before="do_generate_response"
          )
          self.machine.add_transition(
              trigger="no_tool_needed", 
              source="retrieve_knowledge", 
              dest="generate_response",
              before="do_generate_response"
          )
          self.machine.add_transition(
              trigger="finish", 
              source="generate_response", 
              dest="end"
          )
      # 具体动作实现
      def do_retrieve_knowledge(self):
          self.knowledge = self.agent.retrieve_knowledge(self.agent.query)
          print(f"检索到知识：{self.knowledge}")
      
      def do_call_tool(self):
          self.tool_result = self.agent.call_tool(self.agent.query, self.knowledge)
          print(f"工具调用结果：{self.tool_result}")
      
      def do_generate_response(self):
          self.response = self.agent.generate_response(
              self.agent.query, 
              self.knowledge,
              getattr(self, 'tool_result', None)
          )
          print(f"生成回答：{self.response}")
  
  # 使用示例
  class ECommerceAgent:
      def __init__(self, query, kb, tool_registry, llm):
          self.query = query
          self.kb = kb
          self.tool_registry = tool_registry
          self.llm = llm
      
      def retrieve_knowledge(self, query):
          return hybrid_search(query, self.kb)
      
      def call_tool(self, query, knowledge):
          return call_tool_with_llm(query, self.llm, self.tool_registry)
      
      def generate_response(self, query, knowledge, tool_result=None):
          # 整合知识和工具结果生成回答
          return self.llm.generate([f"基于知识{knowledge}和工具结果{tool_result}回答：{query}"])
  
  # 运行工作流
  agent = ECommerceAgent(query="查北京明天天气并推荐适合穿的衣服", ...)
  workflow = AgentWorkflow(agent)
  workflow.start()  # 开始流程
  if "天气" in agent.query:  # 判断是否需要工具
      workflow.needs_tool()
  else:
      workflow.no_tool_needed()
  workflow.finish()
  

• 可视化工作流集成，如果需要与平台的可视化界面兼容，可将工作流定义为 JSON 格式，再通过代码解析执行：

• # 工作流JSON定义（可由前端可视化界面生成）
  workflow_def = {
      "nodes": [
          {"id": "start", "type": "start", "next": "retrieve"},
          {"id": "retrieve", "type": "retrieve_knowledge", "next": "decision"},
          {"id": "decision", "type": "condition", 
              "condition": "contains(query, '天气')",
              "true_next": "call_weather",
              "false_next": "generate"
          },
          {"id": "call_weather", "type": "tool", "tool_name": "weather_query", "next": "generate"},
          {"id": "generate", "type": "generate_response", "next": "end"},
          {"id": "end", "type": "end"}
      ]
  }
  # 解析执行工作流
  def execute_workflow(workflow_def, agent):
      current_node_id = "start"
      while current_node_id != "end":
          node = next(n for n in workflow_def["nodes"] if n["id"] == current_node_id)
          if node["type"] == "retrieve_knowledge":
              agent.retrieve_knowledge()
              current_node_id = node["next"]
          elif node["type"] == "condition":
              # 执行条件判断（实际中可用eval或安全表达式引擎）
              condition_met = "天气" in agent.query
              current_node_id = node["true_next"] if condition_met else node["false_next"]
          # 其他节点类型处理...
  

核心技术栈扩展：向量数据库（FAISS/Chroma/Qdrant）；LLM 框架（LangChain/LlamaIndex）；API 开发（FastAPI/Flask）；工作流引擎（Apache Airflow/Prefect）

实践路线图

• 阶段 1：用 LangChain 复现本文代码，搭建最小知识库 + 工具的智能体
• 阶段 2：集成到 Dify/Coze 平台，对比自建与平台实现的差异
• 阶段 3：针对具体业务场景（如客服 / 数据分析）构建完整解决方案
• 阶段 4：实现资源监控与自动优化（如知识库自动更新、提示词 A/B 测试）

在复杂业务场景中，单一智能体难以处理所有任务，可采用多智能体协同模式：

• 分工模式：按专业领域划分（如财务智能体、法务智能体）
• 通信机制：通过标准化消息格式实现智能体间数据传递
• 调度策略：基于任务复杂度和智能体负载动态分配任务