AI 工作流搭建实战

概述

AI 工作流（AI Workflow）是将大语言模型与业务流程相结合的自动化系统。通过可视化的流程编排工具，无需编写代码即可构建复杂的 AI 应用。

常见工作流类型

Dify 快速入门

环境部署

Docker 部署（推荐）

bashCopy code
1# 克隆仓库
2git clone https://github.com/langgenius/dify.git
3cd dify/docker
4
5# 启动服务
6docker-compose up -d
7
8# 访问 http://localhost/install 完成初始化

本地开发环境

bashCopy code
1# 后端
2cd api
3poetry install
4poetry shell
5flask db upgrade
6flask run --host 0.0.0.0 --port 5001
7
8# 前端
9cd web
10npm install
11npm run dev

第一个工作流：智能客服

步骤 1: 创建应用

1. 登录 Dify 控制台
2. 点击「创建应用」→「工作流编排」
3. 命名：智能客服助手

步骤 2: 设计流程

步骤 3: 配置节点

意图识别节点配置：

jsonCopy code
1{
2  "model": "gpt-4",
3  "temperature": 0.3,
4  "system_prompt": "你是一个客服意图识别专家。请分析用户问题，判断意图类型：\n1. technical - 技术问题\n2. product - 产品咨询\n3. complaint - 投诉建议\n4. other - 其他\n\n只输出意图类型，不要解释。",
5  "user_prompt": "用户问题：{{#start.user_input#}}"
6}

知识检索节点配置：

jsonCopy code
1{
2  "knowledge_base": "产品知识库",
3  "retrieval_mode": "semantic",
4  "top_k": 3,
5  "score_threshold": 0.7
6}

答案生成节点配置：

jsonCopy code
1{
2  "model": "gpt-4",
3  "temperature": 0.7,
4  "system_prompt": "你是专业客服助手。基于检索到的知识回答用户问题。如果知识不足，礼貌地告知用户需要转人工。",
5  "user_prompt": "用户问题：{{#start.user_input#}}\n\n相关知识：{{#retrieval.result#}}"
6}

进阶工作流：内容生成流水线

场景描述

自动生成一篇完整的公众号文章：选题 → 大纲 → 正文 → 配图建议 → 标题优化

流程设计

yamlCopy code
1workflow:
2  name: 公众号文章生成器
3  
4  nodes:
5    - id: start
6      type: start
7      inputs:
8        - name: topic
9          type: text
10          label: 文章主题
11    
12    - id: research
13      type: http-request
14      config:
15        url: "https://api.example.com/trending"
16        method: GET
17        params:
18          keyword: "{{#start.topic#}}"
19      
20    - id: outline
21      type: llm
22      config:
23        model: claude-3-opus
24        prompt: |
25          基于主题「{{#start.topic#}}」和热点数据：
26          {{#research.result#}}
27          
28          生成文章大纲，包含：
29          1. 吸引人的标题（3个备选）
30          2. 文章结构（引言-正文-结论）
31          3. 每个部分的核心观点
32    
33    - id: content
34      type: llm
35      config:
36        model: gpt-4
37        prompt: |
38          根据大纲撰写完整文章：
39          {{#outline.result#}}
40          
41          要求：
42          - 字数：1500-2000字
43          - 风格：专业但易懂
44          - 包含具体案例和数据
45    
46    - id: image-suggestions
47      type: llm
48      config:
49        model: gpt-4-vision
50        prompt: |
51          为以下文章内容推荐配图：
52          {{#content.result#}}
53          
54          每张图需说明：
55          - 图片主题
56          - 构图建议
57          - 配色方案
58    
59    - id: title-optimize
60      type: llm
61      config:
62        model: claude-3-haiku
63        prompt: |
64          优化以下标题，生成10个爆款标题选项：
65          {{#outline.titles#}}
66          
67          优化方向：
68          - 数字法则
69          - 悬念设置
70          - 情感共鸣
71          - 实用性强调
72    
73    - id: end
74      type: end
75      outputs:
76        - name: article
77          value: "{{#content.result#}}"
78        - name: titles
79          value: "{{#title-optimize.result#}}"
80        - name: images
81          value: "{{#image-suggestions.result#}}"

LangFlow 工作流搭建

安装部署

bashCopy code
1pip install langflow
2
3# 启动
4langflow run
5
6# 访问 http://localhost:7860

搭建数据提取工作流

场景： 从合同 PDF 中提取关键信息

组件清单：

1. File - 上传 PDF 文件
2. PDF Loader - 解析文档内容
3. Split Text - 文档分块
4. ChatOpenAI - LLM 处理
5. Prompt - 提取提示词
6. Output Parser - JSON 格式化

Prompt 配置：

jsonCopy code
1从以下合同内容中提取信息，以 JSON 格式返回：
2
3合同内容：
4{document}
5
6提取字段：
7- 合同编号 (contract_no)
8- 签约日期 (sign_date)
9- 甲方名称 (party_a)
10- 乙方名称 (party_b)
11- 合同金额 (amount)
12- 付款方式 (payment_terms)
13- 服务期限 (service_period)
14- 违约责任 (liability)
15
16只返回 JSON，不要其他内容。

智能体（Agent）工作流

ReAct 模式实现

ReAct（Reasoning + Acting）让 AI 能够思考并调用工具：

pythonCopy code
1from langchain.agents import Tool, AgentExecutor, create_react_agent
2from langchain.prompts import PromptTemplate
3from langchain.tools import DuckDuckGoSearchRun
4
5# 定义工具
6tools = [
7    Tool(
8        name="搜索",
9        func=DuckDuckGoSearchRun().run,
10        description="用于搜索实时信息"
11    ),
12    Tool(
13        name="计算器",
14        func=lambda x: eval(x),
15        description="用于数学计算"
16    ),
17    Tool(
18        name="数据库查询",
19        func=query_database,
20        description="用于查询企业数据库"
21    )
22]
23
24# ReAct Prompt
25template = """尽可能回答以下问题。你可以使用以下工具：
26
27{tools}
28
29请使用以下格式：
30
31问题：你需要回答的问题
32思考：你应该如何思考
33行动：要采取的行动（必须是以下工具之一：{tool_names}）
34行动输入：工具的输入
35观察：行动的结果
36...（这个思考/行动/行动输入/观察可以重复N次）
37思考：我现在知道最终答案
38最终答案：问题的最终答案
39
40开始！
41
42问题：{input}
43思考：{agent_scratchpad}"""
44
45prompt = PromptTemplate.from_template(template)
46
47# 创建 Agent
48agent = create_react_agent(llm, tools, prompt)
49agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
50
51# 运行
52result = agent_executor.invoke({
53    "input": "帮我查一下今天北京的天气，然后计算一下如果温度是25度，华氏度是多少？"
54})

多 Agent 协作系统

pythonCopy code
1from langchain.agents import AgentType, initialize_agent
2from langchain.schema import SystemMessage
3
4# 研究员 Agent
5researcher = initialize_agent(
6    tools=[search_tool, arxiv_tool],
7    llm=llm,
8    agent=AgentType.OPENAI_FUNCTIONS,
9    system_message=SystemMessage(content="你是研究员，负责收集和整理信息。"),
10    verbose=True
11)
12
13# 写手 Agent
14writer = initialize_agent(
15    tools=[doc_tool],
16    llm=llm,
17    agent=AgentType.OPENAI_FUNCTIONS,
18    system_message=SystemMessage(content="你是专业写手，负责撰写高质量文章。"),
19    verbose=True
20)
21
22# 编辑 Agent
23editor = initialize_agent(
24    tools=[grammar_tool],
25    llm=llm,
26    agent=AgentType.OPENAI_FUNCTIONS,
27    system_message=SystemMessage(content="你是资深编辑，负责审核和修改文章。"),
28    verbose=True
29)
30
31# 协作流程
32def collaborative_workflow(topic):
33    # 1. 研究
34    research = researcher.run(f"搜集关于'{topic}'的最新资料和数据")
35    
36    # 2. 写作
37    draft = writer.run(f"基于以下资料撰写文章：\n{research}")
38    
39    # 3. 编辑
40    final = editor.run(f"审核并优化以下文章：\n{draft}")
41    
42    return final

工作流编排最佳实践

1. 错误处理

yamlCopy code
1- id: main_task
2  type: llm
3  fallback:
4    - id: fallback_model
5      type: llm
6      config:
7        model: gpt-3.5-turbo  # 降级模型
8  on_error: continue  # 继续执行后续节点

2. 并行处理

yamlCopy code
1- id: parallel_analysis
2  type: parallel
3  branches:
4    - id: sentiment
5      type: llm
6      config:
7        prompt: "分析情感：{{#input.text#}}"
8    - id: keywords
9      type: llm
10      config:
11        prompt: "提取关键词：{{#input.text#}}"
12    - id: summary
13      type: llm
14      config:
15        prompt: "生成摘要：{{#input.text#}}"
16  merge_strategy: all  # 等待所有分支完成

3. 人工审核节点

yamlCopy code
1- id: human_review
2  type: human-in-the-loop
3  config:
4    assignee: "manager"
5    timeout: 3600  # 1小时超时
6    fallback_action: "reject"
7  inputs:
8    - name: content
9      value: "{{#llm.output#}}"

4. 缓存策略

pythonCopy code
1# 工作流级别缓存
2@workflow.cache(ttl=3600)
3def expensive_operation(input_data):
4    # 耗时操作
5    return result
6
7# 节点级别缓存
8- id: cached_node
9  type: llm
10  cache:
11    enabled: true
12    key: "{{#input.topic#}}"
13    ttl: 7200

性能优化

1. 模型选择策略

pythonCopy code
1# 简单任务使用轻量级模型
2if task_complexity < 0.5:
3    model = "gpt-3.5-turbo"
4else:
5    model = "gpt-4"
6
7# 动态温度设置
8if task_type == "creative":
9    temperature = 0.8
10else:
11    temperature = 0.2

2. 流式输出

pythonCopy code
1# Dify API 流式调用
2import requests
3
4def stream_workflow(query):
5    response = requests.post(
6        "http://localhost/v1/workflows/run",
7        json={
8            "inputs": {"query": query},
9            "response_mode": "streaming",
10            "user": "user-123"
11        },
12        stream=True
13    )
14    
15    for line in response.iter_lines():
16        if line:
17            yield json.loads(line)
18
19# 使用
20for chunk in stream_workflow("帮我写一首诗"):
21    print(chunk.get("answer", ""), end="")

3. 批量处理

pythonCopy code
1from concurrent.futures import ThreadPoolExecutor
2
3def batch_process(items, workflow, max_workers=5):
4    with ThreadPoolExecutor(max_workers=max_workers) as executor:
5        futures = [
6            executor.submit(workflow.run, item)
7            for item in items
8        ]
9        results = [f.result() for f in futures]
10    return results

监控与调试

1. 日志追踪

pythonCopy code
1import logging
2
3logging.basicConfig(
4    level=logging.INFO,
5    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
6)
7
8logger = logging.getLogger(__name__)
9
10@workflow.trace()
11def my_node(inputs):
12    logger.info(f"Processing inputs: {inputs}")
13    result = process(inputs)
14    logger.info(f"Generated output: {result}")
15    return result

2. 可视化调试

Dify 和 LangFlow 都提供可视化调试功能：

• 节点状态查看: 每个节点的输入输出
• 执行时间分析: 识别性能瓶颈
• 变量追踪: 查看数据流变化
• 重试运行: 从失败节点重新执行

总结

AI 工作流是实现业务自动化的有效方式，通过可视化的编排工具，非技术人员也能构建强大的 AI 应用。关键要点：

1. 模块化设计: 将复杂流程拆分为独立节点
2. 错误处理: 设计降级策略和重试机制
3. 性能优化: 合理选择模型，使用缓存和并行
4. 持续迭代: 基于反馈不断优化提示词和流程

如需进一步支持，请联系我们：

• 邮箱：c@m9ai.work
• 官网：https://m9ai.work