摘要

本文基于 AI Agent 工作流平台化思路，拆解“工具发现、智能体编排、流程自动化”的核心架构，并以客户反馈分析场景为例，演示如何用 Python 调用大模型 API 构建可重复运行的业务自动化流程。

---

背景介绍：AI Agent 的难点不在“聊天”，而在“流程闭环”

当前很多团队在尝试落地 AI Agent 时，容易将重点放在“让模型回答问题”上。但在真实业务系统中，Agent 只是其中一个环节。一个可用的 AI Agent 系统通常至少包含以下能力：

1. 合适的模型：需要具备较强的推理、总结、结构化输出能力。
2. 业务上下文：如产品文档、知识库、历史工单、CRM 数据等。
3. 工具调用能力：能够访问文档、搜索信息、调用 API、发送通知。
4. 多步骤编排：将数据采集、分析、比对、总结、分发串联起来。
5. 自动化触发：支持定时任务、Webhook、API 事件触发，而不是人工每天执行。

视频中提到的核心观点非常关键：AI Agent 平台的价值不只是提供一个聊天界面，而是提供一个用于构建业务工作流的工作空间。其典型链路可以概括为：

Agent Marketplace → Playground → Flow Builder

也就是：

• 在市场中找到可用的 Agent 和工具；
• 在 Playground 中组合成面向业务场景的工作流；
• 通过 Flow Builder 将工作流转成可重复执行的自动化流程。

这个思路对于企业 AI 应用开发非常有参考价值。

---

核心原理：从单模型调用到多智能体工作流

1. Agent Marketplace：工具与能力的标准化

在传统开发中，如果要实现一个“客户反馈日报”系统，通常需要开发者分别处理：

• 工单系统 API；
• 知识库检索；
• 文本聚类与主题归纳；
• 大模型摘要；
• Slack / 邮件 / Webhook 通知；
• 定时任务调度。

如果每个团队成员都用不同脚本实现，系统很快会变得不可维护。

Agent Marketplace 的本质是将这些能力封装成标准化组件，例如：

• 文档解析 Agent；
• 知识库检索 Agent；
• 反馈分类 Agent；
• 摘要生成 Agent；
• 通知分发 Agent。

这样开发者不需要从零实现所有能力，而是围绕业务目标进行组合。

2. Playground：多智能体编排

Playground 更像是工作流实验环境。以“客户反馈分析”为例，可以拆成多个职责明确的 Agent：

• Feedback Collector：收集新增反馈；
• Knowledge Retriever：检索产品文档或知识库；
• Issue Analyzer：识别高频问题、严重程度和影响范围；
• Summary Writer：生成面向团队的摘要；
• Notifier：将结果发送到团队协作系统。

这种方式比“把所有需求塞给一个模型”更稳定，因为每个步骤职责更清晰，也更容易调试。

3. Flow Builder：可重复执行的自动化

真正的业务价值来自自动化。一个工作流如果每天还需要人工打开页面点击运行，本质上仍然是半自动。

Flow Builder 的核心能力包括：

• 定时触发，例如每天 9 点运行；
• Webhook 触发，例如新工单进入时运行；
• API 触发，例如外部系统主动调用；
• 可视化流程，便于团队理解和审计；
• 失败重试、日志记录、结果追踪。

这也是 AI Agent 从 Demo 走向生产系统的关键。

---

技术资源与工具选型

在模型接入层，我个人常用的是 薛定猫AI（xuedingmao.com）。它采用 OpenAI 兼容接口，开发时只需要配置 base_url + api_key + model，即可切换不同模型。

从工程角度看，它的价值主要体现在：

• 聚合 500+ 主流大模型，包括 GPT-5.4、Claude 4.6、Gemini 3.1 Pro 等；
• 新模型实时首发，开发者可以更早体验前沿模型 API；
• 统一接入接口，降低多模型集成复杂度；
• 对多 Agent 系统比较友好，方便在不同任务中切换不同模型。

下面示例中默认使用 claude-opus-4-6。该模型适合复杂推理、长上下文分析、结构化总结等场景，在客户反馈归因、知识库比对、多步骤摘要生成中表现较强。

---

实战演示：用 Python 构建客户反馈 AI 工作流

下面实现一个简化版的客户反馈自动化流程：

1. 读取客户反馈；
2. 检索内部知识库；
3. 调用大模型分析高频问题；
4. 生成结构化日报；
5. 可选发送到 Webhook。

完整代码示例

import os
import json
import requests
from typing import List, Dict, Any


# =========================
# 基础配置
# =========================

XUEDINGMAO_API_KEY = os.getenv("XUEDINGMAO_API_KEY", "替换为你的API_KEY")
BASE_URL = "https://xuedingmao.com/v1"
MODEL_NAME = "claude-opus-4-6"

# 可选：企业微信、飞书、Slack 或自建系统 Webhook
WEBHOOK_URL = os.getenv("REPORT_WEBHOOK_URL", "")


# =========================
# 示例业务数据
# =========================

CUSTOMER_FEEDBACK = [
    {
        "user": "客户A",
        "content": "最近导出报表特别慢，超过 2 万行数据时经常卡住。",
        "channel": "support_ticket"
    },
    {
        "user": "客户B",
        "content": "权限配置页面不太清楚，管理员不知道如何给销售人员开放报表权限。",
        "channel": "email"
    },
    {
        "user": "客户C",
        "content": "批量导入客户数据时，如果 Excel 中有空行，系统没有明确错误提示。",
        "channel": "support_ticket"
    },
    {
        "user": "客户D",
        "content": "报表导出等待时间很长，希望能支持后台异步导出并通知。",
        "channel": "chat"
    }
]

INTERNAL_KNOWLEDGE_BASE = [
    {
        "title": "报表导出性能说明",
        "content": "当前同步导出适合 1 万行以内数据。超过 1 万行建议使用异步导出任务。"
    },
    {
        "title": "权限管理文档",
        "content": "管理员可在系统设置-角色权限中，为不同角色配置报表查看、导出和编辑权限。"
    },
    {
        "title": "数据导入规则",
        "content": "Excel 导入时，系统会校验必填字段、格式和重复客户，但空行提示仍在优化中。"
    }
]


# =========================
# 工具函数：简单知识库检索
# =========================

def retrieve_related_docs(feedback_items: List[Dict[str, str]],
                          docs: List[Dict[str, str]],
                          top_k: int = 3) -> List[Dict[str, str]]:
    """
    简化版知识库检索：
    生产环境可替换为向量数据库，如 Milvus、Qdrant、pgvector、Elasticsearch 等。
    """
    query_text = " ".join(item["content"] for item in feedback_items)
    scored_docs = []

    for doc in docs:
        score = 0
        for token in ["报表", "导出", "权限", "导入", "Excel", "异步", "错误提示"]:
            if token in query_text and token in doc["content"]:
                score += 1

        scored_docs.append({
            "title": doc["title"],
            "content": doc["content"],
            "score": score
        })

    scored_docs.sort(key=lambda x: x["score"], reverse=True)
    return scored_docs[:top_k]


# =========================
# 大模型调用
# =========================

def call_llm(messages: List[Dict[str, str]]) -> str:
    """
    调用 OpenAI 兼容接口。
    薛定猫AI使用 URL + Key + Model 的方式接入，适合多模型统一管理。
    """
    url = f"{BASE_URL}/chat/completions"

    headers = {
        "Authorization": f"Bearer {XUEDINGMAO_API_KEY}",
        "Content-Type": "application/json"
    }

    payload = {
        "model": MODEL_NAME,
        "messages": messages,
        "temperature": 0.2,
        "max_tokens": 2000
    }

    response = requests.post(url, headers=headers, json=payload, timeout=60)

    if response.status_code != 200:
        raise RuntimeError(
            f"LLM API 调用失败，状态码：{response.status_code}，响应：{response.text}"
        )

    data = response.json()
    return data["choices"][0]["message"]["content"]


# =========================
# Agent Step 1：分析客户反馈
# =========================

def analyze_feedback(feedback_items: List[Dict[str, str]],
                     related_docs: List[Dict[str, str]]) -> str:
    feedback_text = json.dumps(feedback_items, ensure_ascii=False, indent=2)
    docs_text = json.dumps(related_docs, ensure_ascii=False, indent=2)

    system_prompt = """
你是一名资深 SaaS 产品运营分析专家，擅长从客户反馈中识别高频问题、影响范围和产品改进方向。
请严格基于输入内容分析，不要编造不存在的信息。
输出要求：
1. 使用 Markdown；
2. 给出问题分类、出现频次、影响评估；
3. 结合内部知识库判断是否已有解决方案；
4. 给出优先级和建议动作。
"""

    user_prompt = f"""
以下是今日新增客户反馈：

{feedback_text}

以下是检索到的内部知识库内容：

{docs_text}

请生成一份面向产品、研发、客服团队的客户反馈分析报告。
"""

    return call_llm([
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_prompt}
    ])


# =========================
# Agent Step 2：发送结果
# =========================

def send_report(report: str) -> None:
    """
    将分析结果发送到外部系统。
    如果未配置 WEBHOOK，则直接打印。
    """
    if not WEBHOOK_URL:
        print("未配置 WEBHOOK_URL，以下为生成的分析报告：\n")
        print(report)
        return

    payload = {
        "msg_type": "text",
        "content": {
            "text": report
        }
    }

    response = requests.post(WEBHOOK_URL, json=payload, timeout=30)

    if response.status_code >= 300:
        raise RuntimeError(f"Webhook 发送失败：{response.status_code}, {response.text}")

    print("分析报告已成功发送。")


# =========================
# Flow Builder：工作流入口
# =========================

def run_customer_feedback_workflow() -> None:
    """
    一个可被定时任务、Webhook 或 API 调用的工作流入口。
    对应可视化 Flow Builder 中的：
    Trigger -> Retrieve Docs -> Analyze -> Notify
    """
    print("开始执行客户反馈 AI 工作流...")

    related_docs = retrieve_related_docs(
        feedback_items=CUSTOMER_FEEDBACK,
        docs=INTERNAL_KNOWLEDGE_BASE
    )

    report = analyze_feedback(
        feedback_items=CUSTOMER_FEEDBACK,
        related_docs=related_docs
    )

    send_report(report)

    print("工作流执行完成。")


if __name__ == "__main__":
    run_customer_feedback_workflow()

运行方式

安装依赖：

pip install requests

配置 API Key：

export XUEDINGMAO_API_KEY="你的API_KEY"

执行脚本：

python feedback_agent_workflow.py

如果需要每天自动运行，可以结合 Linux crontab：

0 9 * * * /usr/bin/python3 /path/to/feedback_agent_workflow.py

这样就可以实现每天上午 9 点自动生成客户反馈分析报告。

---

注意事项：Agent 工作流生产化必须关注这些问题

1. 不要让单个 Agent 承担所有职责

复杂业务流程建议拆分为多个职责单一的 Agent。这样有利于调试、复用和权限控制。

2. 知识库检索要可追溯

生产环境中，知识库检索不应只返回答案，还应返回引用来源、文档版本和命中片段，避免模型生成不可验证内容。

3. 输出结果应结构化

对于下游系统，建议让模型输出 JSON 或固定 Markdown 模板，便于解析、存档和二次处理。

4. 自动化流程需要日志和重试

AI 工作流接入业务系统后，需要记录：

• 输入数据；
• 模型响应；
• 工具调用结果；
• 失败原因；
• 重试次数。

否则一旦流程异常，很难定位问题。

5. 企业场景要重视隐私和权限

涉及客户数据、内部知识库、销售线索时，需要控制数据访问范围。多智能体系统中，不同 Agent 最好只访问完成任务所需的最小数据集。

---

总结

AI Agent 的落地关键不只是模型能力，而是能否形成稳定、可解释、可重复执行的业务工作流。视频中提到的“Marketplace + Playground + Flow Builder”提供了很好的工程化思路：先标准化工具，再编排智能体，最后将流程自动化。

对于开发者而言，可以先从一个高频、重复、规则相对清晰的业务流程切入，例如客户反馈日报、销售线索研究、客服工单分流、运营周报生成等。只要流程边界清晰，AI Agent 就能逐步从辅助工具演进为真正的业务自动化系统。

#AI #大模型 #Python #机器学习 #技术实战