行业 Agent 案例

Agent 在不同行业的应用模式差异显著。本章通过客服、研究和数据分析三个典型场景展示 Agent 的设计与实现。

场景全景图

案例一：智能客服 Agent

"""
智能客服 Agent 实现
"""
from dataclasses import dataclass, field
from enum import Enum
class Intent(Enum):
INQUIRY = "inquiry"         # 咨询
COMPLAINT = "complaint"     # 投诉
REFUND = "refund"           # 退款
TECHNICAL = "technical"     # 技术支持
TRANSFER = "transfer"       # 转人工
UNKNOWN = "unknown"
@dataclass
class CustomerContext:
"""客户上下文"""
customer_id: str
name: str
tier: str = "standard"         # standard / premium / vip
history: list[dict] = field(default_factory=list)
current_intent: Intent = Intent.UNKNOWN
sentiment_score: float = 0.5   # 0=负面, 1=正面
@dataclass
class AgentResponse:
"""Agent 回复"""
message: str
action: str | None = None      # 触发的动作
escalate: bool = False         # 是否转人工
confidence: float = 0.0
class CustomerServiceAgent:
"""客服 Agent"""
# 自动处理能力边界
AUTO_HANDLE = {Intent.INQUIRY, Intent.TECHNICAL}
NEED_CONFIRM = {Intent.REFUND}
ALWAYS_ESCALATE = {Intent.COMPLAINT}
ESCALATION_SENTIMENT_THRESHOLD = 0.3  # 情绪低于此值转人工
def __init__(self, knowledge_base, order_system):
self.kb = knowledge_base
self.orders = order_system
def handle(self, message: str, context: CustomerContext) -> AgentResponse:
"""处理客户消息"""
# 1. 意图识别
intent = self._classify_intent(message)
context.current_intent = intent
# 2. 情绪检测
sentiment = self._detect_sentiment(message)
context.sentiment_score = sentiment
# 3. 路由决策
if intent in self.ALWAYS_ESCALATE:
return AgentResponse(
message="非常抱歉给您带来不好的体验，我正在为您转接专属客服...",
action="escalate_to_human",
escalate=True,
)
if sentiment < self.ESCALATION_SENTIMENT_THRESHOLD:
return AgentResponse(
message="我理解您的心情，让我为您转接高级客服来处理...",
action="escalate_to_senior",
escalate=True,
)
# 4. 处理意图
if intent == Intent.INQUIRY:
return self._handle_inquiry(message, context)
elif intent == Intent.REFUND:
return self._handle_refund(message, context)
elif intent == Intent.TECHNICAL:
return self._handle_technical(message, context)
else:
return AgentResponse(
message="请问您需要什么帮助？我可以处理咨询、退款和技术支持相关问题。",
confidence=0.3,
)
def _classify_intent(self, message: str) -> Intent:
"""意图分类（简化实现）"""
keywords = {
Intent.REFUND: ["退款", "退货", "退钱"],
Intent.COMPLAINT: ["投诉", "举报", "差评"],
Intent.TECHNICAL: ["报错", "bug", "无法", "故障"],
Intent.TRANSFER: ["人工", "转接"],
}
for intent, words in keywords.items():
if any(w in message for w in words):
return intent
return Intent.INQUIRY
def _detect_sentiment(self, message: str) -> float:
"""情绪检测（简化实现）"""
negative = ["生气", "愤怒", "垃圾", "太差", "骗人"]
if any(w in message for w in negative):
return 0.2
return 0.7
def _handle_inquiry(self, message: str, context: CustomerContext) -> AgentResponse:
result = self.kb.search(message)
return AgentResponse(message=result, confidence=0.8)
def _handle_refund(self, message: str, context: CustomerContext) -> AgentResponse:
return AgentResponse(
message="我已查看您的订单，退款申请需要确认。请问您的订单号是？",
action="initiate_refund_flow",
confidence=0.7,
)
def _handle_technical(self, message: str, context: CustomerContext) -> AgentResponse:
result = self.kb.search(f"技术: {message}")
return AgentResponse(message=result, confidence=0.75)

案例二：研究分析 Agent

"""
多步骤研究分析 Agent
"""
from dataclasses import dataclass, field
@dataclass
class ResearchTask:
"""研究任务"""
topic: str
depth: str = "standard"    # brief / standard / deep
sources_limit: int = 10
language: str = "zh"
@dataclass
class Source:
"""信息源"""
title: str
url: str
snippet: str
relevance: float
@dataclass
class ResearchReport:
"""研究报告"""
topic: str
summary: str
key_findings: list[str]
sources: list[Source]
recommendations: list[str]
class ResearchAgent:
"""研究 Agent"""
def __init__(self, search_tool, llm_client):
self.search = search_tool
self.llm = llm_client
def research(self, task: ResearchTask) -> ResearchReport:
"""执行完整研究流程"""
# Step 1: 分解研究问题
sub_questions = self._decompose_topic(task.topic)
print(f"  分解为 {len(sub_questions)} 个子问题")
# Step 2: 多轮搜索
all_sources: list[Source] = []
for q in sub_questions:
sources = self._search_and_filter(q, limit=task.sources_limit)
all_sources.extend(sources)
# Step 3: 去重和排序
unique_sources = self._deduplicate(all_sources)
top_sources = sorted(unique_sources, key=lambda s: s.relevance, reverse=True)[
: task.sources_limit
]
# Step 4: 综合分析
findings = self._analyze_sources(task.topic, top_sources)
# Step 5: 生成报告
report = ResearchReport(
topic=task.topic,
summary=findings["summary"],
key_findings=findings["key_points"],
sources=top_sources,
recommendations=findings["recommendations"],
)
return report
def _decompose_topic(self, topic: str) -> list[str]:
"""将研究主题分解为子问题"""
prompt = f"将以下研究主题分解为 3-5 个具体的搜索问题：\n{topic}"
result = self.llm.generate(prompt)
return [line.strip("- ") for line in result.strip().split("\n") if line.strip()]
def _search_and_filter(self, query: str, limit: int = 5) -> list[Source]:
"""搜索并过滤结果"""
raw = self.search.search(query, limit=limit)
return [
Source(
title=r["title"],
url=r["url"],
snippet=r["snippet"],
relevance=r.get("score", 0.5),
)
for r in raw
]
def _deduplicate(self, sources: list[Source]) -> list[Source]:
"""去重"""
seen_urls: set[str] = set()
unique = []
for s in sources:
if s.url not in seen_urls:
seen_urls.add(s.url)
unique.append(s)
return unique
def _analyze_sources(self, topic: str, sources: list[Source]) -> dict:
"""综合分析所有信息源"""
ctx = "\n".join(f"- [{s.title}]: {s.snippet}" for s in sources[:10])
prompt = f"基于以下资料，对「{topic}」进行综合分析：\n{ctx}"
result = self.llm.generate(prompt)
return {
"summary": result,
"key_points": [f"发现 {i+1}" for i in range(3)],
"recommendations": ["建议深入研究", "建议关注趋势"],
}

案例三：数据分析 Agent

"""
自然语言数据分析 Agent
"""
from dataclasses import dataclass
@dataclass
class QueryResult:
"""查询结果"""
sql: str
columns: list[str]
rows: list[list]
row_count: int
insight: str = ""
class DataAnalysisAgent:
"""数据分析 Agent"""
# 安全：只允许 SELECT 操作
ALLOWED_OPERATIONS = {"SELECT"}
def __init__(self, db_connection, llm_client):
self.db = db_connection
self.llm = llm_client
def query(self, natural_language: str) -> QueryResult:
"""自然语言查询"""
# 1. NL → SQL
sql = self._text_to_sql(natural_language)
# 2. 安全检查
if not self._validate_sql(sql):
return QueryResult(
sql=sql, columns=[], rows=[], row_count=0,
insight="查询被安全策略拒绝：仅允许 SELECT 操作",
)
# 3. 执行查询
columns, rows = self._execute_sql(sql)
# 4. 生成分析
insight = self._generate_insight(natural_language, columns, rows)
return QueryResult(
sql=sql,
columns=columns,
rows=rows,
row_count=len(rows),
insight=insight,
)
def _text_to_sql(self, text: str) -> str:
"""自然语言转 SQL"""
schema_info = self.db.get_schema_description()
prompt = (
f"根据以下数据库结构，将自然语言转为 SQL：\n"
f"结构：{schema_info}\n问题：{text}\n"
f"只输出 SQL，不需要解释。"
)
return self.llm.generate(prompt).strip()
def _validate_sql(self, sql: str) -> bool:
"""验证 SQL 安全性"""
first_word = sql.strip().split()[0].upper() if sql.strip() else ""
return first_word in self.ALLOWED_OPERATIONS
def _execute_sql(self, sql: str) -> tuple[list[str], list[list]]:
"""执行 SQL"""
return self.db.execute(sql)
def _generate_insight(
self, question: str, columns: list[str], rows: list[list]
) -> str:
"""生成数据洞察"""
preview = rows[:5]
prompt = (
f"问题：{question}\n"
f"列：{columns}\n"
f"数据预览（前5行）：{preview}\n"
f"总行数：{len(rows)}\n"
f"请用简洁的中文总结数据洞察。"
)
return self.llm.generate(prompt)

行业应用选型

本章小结

下一章：Agent 运维与持续迭代