项目概述
传统的 RPA（机器人流程自动化）需要预先定义每一步操作，灵活性差且维护成本高。随着 LLM 的 Function Calling 能力成熟，我们可以构建更智能的自动化工作流：LLM 作为"调度大脑"，根据用户意图自主决定调用哪些工具、以什么顺序执行、如何处理异常。这种 Prompt 驱动的方式极大降低了工作流的配置门槛，同时具备了自我纠错和适应新场景的能力。
本案例以一个实际业务场景为例——智能客服与工单处理自动化：用户提出请求后，LLM 自动判断意图、查询知识库、创建工单、发送通知，全程无需人工干预。你将学会如何设计工具接口、编写 Prompt 模板、实现错误处理和重试机制、编排多步骤工作流。
技术架构
系统采用 Agent Loop 架构，核心是一个持续的"思考-行动-观察"循环：
用户输入 -> 意图识别 -> 工具选择 -> 参数填充 -> 执行工具 -> 结果验证 -> 自我纠错 -> 生成回复
详细步骤
步骤 1：工具设计
工具设计的核心原则是：每个工具职责单一、接口清晰、文档详细。LLM 通过工具的 name 和 description 来决定调用哪个工具，通过 parameters schema 来构造参数。description 写得越好，LLM 的工具选择准确率越高。避免一个工具做太多事情。
from pydantic import BaseModel, Field
from typing import Optional
from enum import Enum
class TicketPriority(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
URGENT = "urgent"
# --- 工具 1：查询知识库 ---
search_knowledge_base = {
"type": "function",
"function": {
"name": "search_knowledge_base",
"description": "在产品知识库中搜索相关信息。用于回答产品使用、配置、故障排查等问题。",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词，应简洁精炼"
},
"category": {
"type": "string",
"enum": ["产品使用", "配置指南", "故障排查", "计费问题"],
"description": "问题分类"
}
},
"required": ["query"]
}
}
}
# --- 工具 2：创建工单 ---
class CreateTicketParams(BaseModel):
title: str = Field(description="工单标题，简洁概括问题")
description: str = Field(description="详细的问题描述")
priority: TicketPriority = Field(description="优先级")
customer_id: Optional[str] = Field(default=None, description="客户 ID（如果已提供）")
assigned_team: str = Field(description="分配给哪个团队处理")
create_ticket = {
"type": "function",
"function": {
"name": "create_ticket",
"description": "创建客户支持工单。当知识库中没有相关答案，或问题需要人工处理时调用。",
"parameters": CreateTicketParams.model_json_schema()
}
}
# --- 工具 3：发送通知 ---
send_notification = {
"type": "function",
"function": {
"name": "send_notification",
"description": "向客户或内部团队发送通知（邮件/短信）。",
"parameters": {
"type": "object",
"properties": {
"recipient": {"type": "string", "description": "接收人"},
"channel": {"type": "string", "enum": ["email", "sms"], "description": "通知渠道"},
"message": {"type": "string", "description": "通知内容"},
"ticket_id": {"type": "string", "description": "关联的工单 ID（如有）"}
},
"required": ["recipient", "channel", "message"]
}
}
}
步骤 2：Prompt 模板设计
System Prompt 定义了 Agent 的角色、行为规则和决策边界。好的 System Prompt 应该明确：(1) 你是谁、你能做什么；(2) 决策逻辑（什么时候用哪个工具）；(3) 输出格式要求；(4) 安全边界（不能做什么）。
SYSTEM_PROMPT = """你是一个智能客服助手，负责处理客户的技术支持请求。
## 工作流程
1. 分析用户的问题，判断是否需要查询知识库
2. 如果能从知识库找到答案，直接回答用户
3. 如果知识库中没有相关信息，或问题需要人工介入：
a. 创建工单，分配给合适的团队
b. 向用户确认工单已创建
c. 如果问题紧急，发送内部通知
## 决策规则
- 产品使用/配置问题：先查知识库，找不到再建工单
- 故障排查：查知识库获取临时解决方案，同时建工单跟进
- 计费问题：直接建工单，分配给财务团队
- 紧急问题（影响业务运行）：建 URGENT 工单 + 内部通知
## 输出要求
- 回答要简洁、准确、友好
- 如果创建工单，必须提供工单编号
- 不要编造不确定的信息
## 安全边界
- 不能泄露其他客户的信息
- 不能承诺你无法保证的服务水平
- 不能绕过正常流程直接解决问题"""
步骤 3：错误处理与重试
工作流在执行过程中可能遇到各种异常：工具调用失败、参数不合法、外部服务不可用等。需要设计健壮的错误处理机制：自动重试（指数退避）、参数修正、降级策略。使用 tenacity 库可以优雅地实现重试逻辑。
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import structlog
logger = structlog.get_logger()
class ToolExecutionError(Exception):
"""工具执行失败"""
pass
class MaxRetriesExceeded(Exception):
"""超过最大重试次数"""
pass
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10),
retry=retry_if_exception_type((ToolExecutionError, TimeoutError)),
before_sleep=lambda retry_state: logger.warning(
"工具调用失败，正在重试",
attempt=retry_state.attempt_number,
error=str(retry_state.outcome.exception())
)
)
def execute_tool(tool_name: str, tool_args: dict) -> dict:
"""执行工具调用，带自动重试"""
try:
logger.info("执行工具", tool=tool_name, args=tool_args)
result = available_tools[tool_name](**tool_args)
# 验证返回结果
if not validate_result(result):
raise ToolExecutionError(f"工具 {tool_name} 返回结果异常")
return result
except TypeError as e:
# 参数类型错误，尝试修正
logger.warning("参数错误，尝试修正", error=str(e))
corrected_args = fix_tool_args(tool_name, tool_args)
return execute_tool(tool_name, corrected_args)
except Exception as e:
logger.error("工具执行失败", tool=tool_name, error=str(e))
raise ToolExecutionError(str(e))
步骤 4：工作流编排
Agent Loop 是整个系统的核心调度引擎。它持续运行直到任务完成或达到最大迭代次数。每次迭代：(1) 将对话历史发送给 LLM；(2) LLM 决定调用工具或返回最终回答；(3) 如果调用工具，执行并将结果追加到对话历史；(4) 循环。需要设置最大迭代次数防止无限循环。
from openai import OpenAI
import json
client = OpenAI()
MAX_ITERATIONS = 10
def run_workflow(user_message: str) -> str:
"""执行 Agent 工作流"""
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message}
]
for iteration in range(MAX_ITERATIONS):
logger.info("Agent 迭代", iteration=iteration + 1)
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=[search_knowledge_base, create_ticket, send_notification],
temperature=0
)
msg = response.choices[0].message
messages.append(msg)
# 如果没有工具调用，说明 LLM 给出了最终回答
if not msg.tool_calls:
return msg.content
# 执行所有工具调用
for tool_call in msg.tool_calls:
fn_name = tool_call.function.name
try:
fn_args = json.loads(tool_call.function.arguments)
result = execute_tool(fn_name, fn_args)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False)
})
except MaxRetriesExceeded:
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps({"error": "工具执行失败，请尝试其他方案"})
})
return "抱歉，处理过程中遇到了问题，已为您创建人工工单。"
# 测试
result = run_workflow("我的服务突然无法访问了，急！")
print(result)
步骤 5：应用场景扩展
掌握了基础框架后，可以将工作流扩展到更多业务场景。关键是理解"LLM + 工具"的通用模式：定义问题域的工具集、设计决策 Prompt、实现执行引擎。以下是几个可扩展的应用方向。
数据报表自动化
用户用自然语言描述需求 -> LLM 生成 SQL -> 执行查询 -> 生成图表 -> 发送邮件
DevOps 自动运维
监控告警 -> LLM 分析根因 -> 查询日志 -> 执行修复脚本 -> 通知相关人员
内容发布流水线
原始素材 -> LLM 生成文案 -> 人工审核 -> 自动排版 -> 多平台发布
常见问题与解决方案
Q: LLM 选错了工具怎么办？
优化方向：(1) 改进工具 description，更清晰地描述适用场景；(2) 在 System Prompt 中添加具体的决策示例（Few-shot）；(3) 对于复杂决策，可以先用 LLM 做一次"规划"，再执行具体工具调用。
Q: Agent 循环无限运行怎么办？
必须设置最大迭代次数（建议 5-10 次）。此外，监控每次迭代的工具调用和结果，如果检测到"循环调用同一工具"或"结果没有进展"，主动中断并降级为人工处理。
Q: 如何处理敏感操作？
对于有副作用的操作（如创建工单、发送通知、修改数据），建议添加"确认"环节：LLM 生成操作计划 -> 展示给用户确认 -> 确认后执行。对于高危操作，可以要求人工审批后才能执行。
学习要点总结
- Function Calling 的核心是"LLM 选择工具，代码执行工具"的协作模式
- 工具 description 的质量直接决定 LLM 的决策准确率
- 错误处理和重试是生产环境必备的能力，不能假设所有调用都会成功
- Agent Loop 必须有最大迭代次数保护，防止无限循环
- 对于有副作用的操作，必须加入确认或审批环节