产品评论情感分析Agent并不是简单调用一次大模型给评论打上好评或差评的标签，而是把属性抽取、情感分类、置信度评估和结构化输出串成一个可复用的分析流程。电商平台、应用商店和本地生活服务每天都会产生大量短文本评论，这些评论往往同时包含多个维度的评价，例如一条评论可能说包装很好但物流太慢，如果只输出一个整体情感分数，后续运营就很难定位问题。Agent的价值就在于把评论拆成属性级意见，并为每个属性给出可量化的情感判断和依据。
一、把单次打分升级为属性级分析流程
传统规则系统处理评论情感时通常依赖情感词典和否定词窗口，比如出现喜欢、满意加正分，出现差、慢加负分。这种方式在较规范的短句上能用，但遇到还行吧、没有想象中那么差、价格贵但质量值这样的表达时，规则很容易误判。更重要的是，规则很难区分评论针对的是产品、物流、客服还是价格，最后只能输出一个笼统的情感值。
Agent方案用大模型作为推理核心，将任务拆成两个阶段：先抽取评论中提到的属性词，再判断每个属性词对应的情感极性。输出格式可以设计为JSON数组，每个元素包含property、sentiment、confidence和evidence。这样做有三个好处。第一，业务方能够按属性聚合，例如统计物流负面率、质量好评率。第二，evidence字段保留了原文依据，便于抽查和回溯。第三，confidence可以作为后续人工复核或规则兜底的触发条件。
流程控制上，Agent通常包含输入清洗、批量请求、结果校验、重试和汇总几个步骤。下面这段代码展示了最基础的调用骨架，实际生产环境还需要加入异步、限流和日志。
import json
from typing import Any
def analyze_review(client, review: str) -> dict[str, Any]:
prompt = build_prompt(review)
for attempt in range(3):
raw = client.chat(prompt, temperature=0.2)
try:
parsed = json.loads(raw)
if validate_schema(parsed):
return parsed
except json.JSONDecodeError:
continue
return {"error": "validation_failed", "review": review}
二、Prompt设计与Pydantic约束
提示词的质量直接决定Agent输出的稳定性。系统提示词需要明确角色、任务边界和输出格式，用户提示词则放入待分析评论。为了降低模型自由发挥的空间，可以在提示词中直接给出属性候选集合，例如物流、包装、质量、价格、客服，并要求模型只从这些候选中选择。如果评论没有提到某属性，就不返回该属性。针对隐式表达，提示词需要明确要求模型根据语义判断，而不是只找情感词。
一个实用的系统提示词可以这样写：
SYSTEM_PROMPT = """ 你是一名电商产品评论分析专家。请对用户给出的评论做属性级情感分析。 只关注以下属性：物流、包装、质量、价格、客服。 对每个提到的属性输出一个对象，包含： property: 属性名称 sentiment: 只能是 positive、negative 或 neutral confidence: 0到1之间的小数，表示判断置信度 evidence: 原文中支撑判断的短语，如果没有则写 implicit 如果评论没有提到某属性，不要输出该属性。 直接输出JSON数组，不要添加任何解释。 """
这里没有使用复杂的角色扮演或多轮对话，而是用明确枚举限制输出空间。实际调优时可以给每个属性补充一两个例子，帮助模型区分近似表达。比如价格贵但质量值，应该输出价格negative、质量positive。例子不用太多，通常每个属性给一到两个就能明显减少误判。
光靠提示词还不够，模型偶尔会输出带注释的JSON、字段拼写错误或直接输出一段解释。可以在Python侧用Pydantic定义Schema，解析后强制校验，不通过就进入重试逻辑。下面是一个精简版数据模型：
from pydantic import BaseModel, Field
from typing import Literal
class AspectSentiment(BaseModel):
property: str
sentiment: Literal["positive", "negative", "neutral"]
confidence: float = Field(ge=0, le=1)
evidence: str
class ReviewAnalysis(BaseModel):
aspects: list[AspectSentiment]
解析时先尝试Pydantic的model_validate，如果抛出ValidationError，可以把错误信息拼到重试提示词中，让模型根据错误修正输出。相比粗暴重试，带错误反馈的重试能显著提高第二次成功概率。
三、可靠性保障：重试、置信度阈值与工具兜底
大模型输出具有随机性，即使temperature设为0也可能出现格式错误。Agent的可靠性不能依赖单次调用成功，需要在流程中加入多层防护。第一层是格式校验，用json.loads和Pydantic保证结构正确；第二层是语义校验，检查property是否在允许列表中，sentiment是否在枚举内；第三层是置信度过滤，低于阈值的判断不直接写入最终结果，而是标记为待复核。
重试策略不建议无脑重试，可以设计指数退避和温度调整。例如第一次失败后保持temperature=0.1，第二次失败后升到0.3并追加错误说明，第三次失败后更换模型或返回兜底结果。下面这段逻辑展示了带错误反馈的重试：
def analyze_with_retry(client, review: str) -> dict:
prompt = build_prompt(review)
last_error = ""
for attempt in range(3):
try:
raw = client.chat(prompt + last_error, temperature=0.1 + attempt * 0.1)
data = json.loads(raw)
analysis = ReviewAnalysis.model_validate(data)
return filter_by_confidence(analysis)
except Exception as exc:
last_error = f"\n上一次输出不符合要求：{exc}，请只输出符合JSON Schema的内容。"
return {"aspects": [], "status": "manual_review"}
代码中filter_by_confidence负责把confidence低于0.6的条目单独摘出，可以写入数据库供人工审核，也可以触发规则词典辅助判断。如果Agent面向大规模评论分析，人工审核不可能覆盖全部低置信度样本，可以根据业务容忍度把阈值设低，将高置信度结果直接用于聚合。
工具兜底是Agent能力的延伸。例如当模型判断某条评论涉及产品质量缺陷且confidence很高时，Agent可以调用通知工具创建工单；当评论中出现极端负面词汇且属性为物流时，可以触发物流系统查询包裹状态。工具调用不一定要使用复杂的Function Calling框架，简单的if判断加消息推送就能解决大部分场景。
四、批量处理与成本优化
真实场景中评论量通常很大，逐条同步调用不仅慢，成本也高。可以采用异步并发加批量请求的方式提高吞吐。如果使用OpenAI兼容接口，可以借助asyncio和信号量控制并发，避免触发限流。下面是一个简化版的批量处理示例：
import asyncio
async def analyze_batch(client, reviews: list[str], max_concurrency: int = 5):
semaphore = asyncio.Semaphore(max_concurrency)
async def worker(review: str):
async with semaphore:
return await asyncio.to_thread(analyze_with_retry, client, review)
tasks = [worker(r) for r in reviews]
return await asyncio.gather(*tasks)
成本优化的另一个思路是分层模型。先用规则或轻量分类器过滤掉明显的好评和差评，只把模糊评论交给大模型。比如包含太差、垃圾、非常好、完美等强信号的评论可以用词典直接打标，剩下大约百分之二十到三十的评论再走Agent流程。这样可以在不牺牲太多准确率的前提下大幅降低成本。
缓存也能减少重复计算。产品评论中经常出现完全相同或高度相似的文本，例如客服很好、物流太慢。可以对评论文本做去重，或者对归一化后的短评论使用哈希缓存。需要注意，如果评论包含具体订单号或用户信息，应先做脱敏，避免缓存键携带敏感数据。缓存失效策略可以根据业务更新时间设定，通常每小时或每天淘汰一次就足够。
延迟方面，属性级分析的输出token数量多于简单情感分类，因此单条耗时略高。可以选择较小参数模型并配合结构化输出模式，例如某些推理服务支持JSON模式或函数调用，能减少输出无关文本。如果对实时性要求高，可以预先批量分析头部热评，将结果写入搜索引擎，让用户查询时直接读缓存。
五、用评估集驱动Prompt迭代
情感分析Agent上线后需要持续观察准确率和召回率，不能只看几条示例就认为效果达标。建议从真实评论中抽样构建评估集，每条评论标注属性级情感三元组。评估集至少覆盖显式好评、显式差评、隐式表达、多属性混合、反讽和口语化表达等类型。可以先用200到500条评论做初步标注，后续按周补充误判样本。
评估时按属性维度计算精确率、召回率和F1值，而不是只看整体准确率。模型可能整体情感判断正确，但在属性归属上出错，例如把物流差归到包装上，这种错误在聚合统计时会造成明显偏差。可以把模型输出的property和sentiment与人工标注做严格匹配，只有两者都一致才算正确。下面是一个简单的评估脚本骨架：
def evaluate_predictions(preds: list[ReviewAnalysis], golds: list[ReviewAnalysis]) -> dict:
tp = fp = fn = 0
for pred, gold in zip(preds, golds):
pred_set = {(a.property, a.sentiment) for a in pred.aspects}
gold_set = {(a.property, a.sentiment) for a in gold.aspects}
tp += len(pred_set & gold_set)
fp += len(pred_set - gold_set)
fn += len(gold_set - pred_set)
precision = tp / (tp + fp) if tp + fp else 0
recall = tp / (tp + fn) if tp + fn else 0
f1 = 2 * precision * recall / (precision + recall) if precision + recall else 0
return {"precision": precision, "recall": recall, "f1": f1}
如果发现某类评论持续误判，不要着急堆更多规则，而是回到提示词和评估集。把误判样本按错误模式聚类，例如所有反讽句都被判成正面，可以在提示词中增加反讽识别的要求，或者补充几个反讽例子。调整后重新跑评估集，观察目标指标是否提升，同时确认其他类型没有明显回退。
迭代过程中还可以利用人工反馈数据微调小模型，但需要注意微调成本与收益。如果评论量大且模式相对固定，微调一个轻量分类器来替代大模型做初筛是划算的；如果业务变化频繁，新属性不断出现，则保持提示词驱动的大模型方案更灵活。Agent架构的优势就在于可以把大模型、规则、工具和缓存组合成可替换的模块，根据实际数据逐步优化。