99行代码！教会你用Doris+DeepSeek实现AI舆情分析

import pymysql
from dbutils.pooled_db import PooledDB
import requests
from langchain.agents import initialize_agent, Tool
from langchain_openai import ChatOpenAI

"""
AI舆情分析｜自动投诉处理系统 v1.0
Description: 基于LangChain Agent的智能流程自动化系统，替代硬编码方式处理投诉数据
"""

# 定义Apache Doris数据库查询的Agent Tool
def dorisTool(query):
    POOL_DORIS_OFF = PooledDB(
        user='{user}',password='{password}',host='{host}',port={port},database='{database}',
        charset='utf8',creator=pymysql,blocking=True,maxusage=None,setsession=[],maxconnections=30,mincached=10,ping=0,
    )
    cursor = POOL_DORIS_OFF.connection().cursor()
    cursor.execute(query)
    columns = [col[0] for col in cursor.description]
    rs = [dict(zip(columns, row)) for row in cursor.fetchall()]
    return rs

# 定义发送钉钉消息的Agent Tool
def dingdingTool(msg):
    markdown_message = f"""
           {msg}
           """
    webhook_url = (
        "{webhook_url}")
    headers = {'Content-Type': 'application/json'}
    payload = {
        "msgtype": "markdown",
        "markdown": {
            "title": "投诉数据明细",
            "text": markdown_message
        }
    }
    try:
        response = requests.post(webhook_url, headers=headers, json=payload)
        response.raise_for_status()
        print("钉钉消息推送成功")
    except requests.exceptions.RequestException as e:
        print(f"钉钉消息推送失败: {e}")

# 定义LangChain Agent工具列表
tools = [
    Tool(
        name="执行sql查询doris数据",
        func=dorisTool,
        description="根据传入的doris sql直接执行，返回结果集，返回格式为list"
    ),
    Tool(
        name="推送钉钉消息",
        func=dingdingTool,
        description="根据传入的msg通过钉钉机器人的形式推送msg到钉钉群"
    )
]

# 初始化LLM
llm = ChatOpenAI(base_url='https://ark.cn-beijing.volces.com/api/v3',
                 api_key="{api_key}",
                 model='{model}')

# 初始化LangChain Agent
agent = initialize_agent(
    tools,
    llm,
    agent="zero-shot-react-description",
    verbose=True,
    handle_parsing_errors=True
)

# 提示模板，指导AI代理如何处理查询
question = "请问近30天有关于一臻数据的投诉信息吗"
promptTemplate = """
你是“一臻数据”平台的投诉识别以及处理员，需完成以下任务：
1. 根据用户的问题，生成一个 SQL 查询，具体查询时间范围以用用户实际问题为准。
2. 从查询结果中筛选出与“一臻数据”相关的投诉信息。
3. 将筛选结果汇总为易读的格式，并给出处理建议以及将命中的投诉信息根据你的建议排出一个紧急程度来，尽可能多的挖掘一些信息返回给客户，做专家指导，如“舆情分类”、“定义分类”、“紧急程度”等。
4. 如果存在与“一臻数据”相关的投诉，将汇总信息和你的处理建议通过钉钉机器人推送到指定群组。
5. 在推送内容中添加表情符号，增强可读性。
注意：
-如需调用工具，根据工具的作用在该任务里调用合适的工具，并根据工具的要求传入需要的参数，要求参数传入格式准确
-只能从获得的doris库中的数据中做分析，不允许自己fake数据。
数据库信息：
- 表名：yz_complaint_detail
- 字段：complaint_time（投诉时间）, detail（具体投诉内容）
- SQL 查询示例：SELECT detail FROM yz_complaint_detail WHERE complaint_time >= NOW() - INTERVAL 60 DAY;
"""
prompt = f"{question}\n{promptTemplate}"

# 运行Agent
try:
    result = agent.run(prompt)
    print("Final Result:")
    print(result)
except Exception as e:
    print(f"代理执行出错: {e}")

-- 测试表
CREATETABLE`yz_complaint_detail` ( 
`complaint_time` datetime NULLCOMMENT'投诉时间',
`detail`textNULLCOMMENT'具体投诉内容',
INDEX index_complaint_time (`complaint_time`) USING INVERTED ) 
ENGINE = OLAP DUPLICATEKEY(`complaint_time`) 
COMMENT'一臻数据投诉信息表'
DISTRIBUTEDBYHASH(`complaint_time`) BUCKETS 30
PROPERTIES ( 
"replication_allocation" = "tag.location.default: 1"
);

-- 测试数据
insertinto yz_complaint_detail values
('2025-03-11 15:30:30','一臻数据真垃圾，总是拖稿'),
('2025-03-10 13:30:30','一臻数据真不错'),
('2025-03-09 15:30:30','一臻数据真垃圾，严重拖稿'),
('2025-02-08 15:30:30','一臻数据服务很好'),
('2025-02-07 15:30:30','一臻数据不处理问题，客服被投诉！'),
('2025-02-06 15:30:30','一臻数据断更被暴利催更'),
('2025-02-05 15:30:30','投诉一臻数据，严重拖稿'),
('2025-03-00 18:30:30','一臻数据真好')