AI Agent 垂直应用场景
不同领域对 Agent 有不同的需求和特点。
通过领域定制,Agent 能够提供更专业的服务。
代码 Agent
代码 Agent 是当前最成功的 Agent 应用领域之一。
能够理解代码库、编写新代码、调试问题。
核心能力
代码补全:根据上下文预测下一段代码。
代码生成:根据自然语言描述生成代码。
代码审查:发现代码中的问题和改进点。
Bug 定位:分析错误信息,定位问题根源。
代码重构:提出并执行代码优化建议。
代表系统
| 系统 | 开发公司 | 特点 |
|---|---|---|
| GitHub Copilot | GitHub/OpenAI | 代码补全为主,实时建议 |
| Claude Code | Anthropic | 命令行助手,深度代码理解 |
| Devin | Cognition | 自主编程,端到端任务执行 |
| Cursor | Cursor | AI 代码编辑器,深度 IDE 集成 |
代码实现示例
Text2SQL Agent 实现
class Text2SQLAgent:
"""
Text2SQL Agent
将自然语言转换为 SQL 查询
"""
def __init__(self, llm, schema):
self.llm = llm
# 数据库 Schema 信息
self.schema = schema
def convert(self, question):
"""
将自然语言问题转换为 SQL
:param question: 自然语言问题
:return: SQL 查询语句
"""
prompt = f"""
你是一个 SQL 专家。
数据库 Schema:
{self.schema}
规则:
1. 只生成 SELECT 查询(不支持 INSERT/UPDATE/DELETE)
2. 使用合理的表别名
3. 添加必要的 JOIN 条件
4. 使用清晰的列别名
问题:{question}
请生成对应的 SQL 查询。
"""
sql = self.llm.generate(prompt)
# 安全检查
return self.sanitize(sql)
def sanitize(self, sql):
"""
SQL 安全检查
确保不包含危险操作
"""
# 转小写检查
sql_lower = sql.lower().strip()
# 检查是否只包含 SELECT
forbidden_keywords = [
"insert", "update", "delete", "drop",
"create", "alter", "truncate", "exec",
"execute", "grant", "revoke"
]
for keyword in forbidden_keywords:
if keyword in sql_lower:
raise ValueError(f"禁止的关键字: {keyword}")
return sql
def execute(self, question, db_connection):
"""
执行 Text2SQL 查询
"""
sql = self.convert(question)
cursor = db_connection.cursor()
cursor.execute(sql)
results = cursor.fetchall()
columns = [desc[0] for desc in cursor.description]
return {"columns": columns, "rows": results}
"""
Text2SQL Agent
将自然语言转换为 SQL 查询
"""
def __init__(self, llm, schema):
self.llm = llm
# 数据库 Schema 信息
self.schema = schema
def convert(self, question):
"""
将自然语言问题转换为 SQL
:param question: 自然语言问题
:return: SQL 查询语句
"""
prompt = f"""
你是一个 SQL 专家。
数据库 Schema:
{self.schema}
规则:
1. 只生成 SELECT 查询(不支持 INSERT/UPDATE/DELETE)
2. 使用合理的表别名
3. 添加必要的 JOIN 条件
4. 使用清晰的列别名
问题:{question}
请生成对应的 SQL 查询。
"""
sql = self.llm.generate(prompt)
# 安全检查
return self.sanitize(sql)
def sanitize(self, sql):
"""
SQL 安全检查
确保不包含危险操作
"""
# 转小写检查
sql_lower = sql.lower().strip()
# 检查是否只包含 SELECT
forbidden_keywords = [
"insert", "update", "delete", "drop",
"create", "alter", "truncate", "exec",
"execute", "grant", "revoke"
]
for keyword in forbidden_keywords:
if keyword in sql_lower:
raise ValueError(f"禁止的关键字: {keyword}")
return sql
def execute(self, question, db_connection):
"""
执行 Text2SQL 查询
"""
sql = self.convert(question)
cursor = db_connection.cursor()
cursor.execute(sql)
results = cursor.fetchall()
columns = [desc[0] for desc in cursor.description]
return {"columns": columns, "rows": results}
数据分析 Agent
数据分析 Agent 能够理解数据需求,执行查询,生成分析报告。
让非技术人员也能进行复杂的数据分析。
核心能力
数据理解:理解数据库结构和数据含义。
查询生成:Text2SQL,将自然语言转为查询。
可视化:生成图表建议和配置。
报告生成:分析结果,生成自然语言报告。
代码实现
数据分析 Agent 实现
class DataAnalysisAgent:
"""
数据分析 Agent
理解数据需求,执行分析,生成报告
"""
def __init__(self, llm, db_connection, chart_generator):
self.llm = llm
self.db = db_connection
self.chart_generator = chart_generator
def analyze(self, request):
"""
处理数据分析请求
"""
# 第一步:理解分析需求
intent = self.parse_intent(request)
# 第二步:生成查询
query = self.generate_query(intent)
# 第三步:执行查询
data = self.execute_query(query)
# 第四步:分析数据
analysis = self.perform_analysis(data, intent)
# 第五步:生成可视化建议
charts = self.suggest_charts(analysis)
# 第六步:生成报告
report = self.generate_report(analysis, charts)
return {
"intent": intent,
"query": query,
"data": data,
"analysis": analysis,
"charts": charts,
"report": report
}
def parse_intent(self, request):
"""解析用户意图"""
prompt = f"""
分析以下数据请求,提取关键信息:
请求:{request}
请提取:
1. 分析目标(比较、趋势、分布等)
2. 涉及的表和字段
3. 时间范围(如果有)
4. 筛选条件(如果有)
"""
return self.llm.generate(prompt)
def generate_query(self, intent):
"""生成 SQL 查询"""
prompt = f"""
基于以下意图生成 SQL 查询:
{intent}
数据库 Schema:
{self.get_schema()}
"""
return self.llm.generate(prompt)
def execute_query(self, query):
"""执行查询"""
cursor = self.db.cursor()
cursor.execute(query)
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
return {"columns": columns, "rows": rows}
def perform_analysis(self, data, intent):
"""执行数据分析"""
# 简化实现
import statistics
rows = data["rows"]
columns = data["columns"]
analysis = {
"row_count": len(rows),
"summary": {}
}
# 对数值列计算统计信息
for i, col in enumerate(columns):
if self.is_numeric(rows, i):
values = [row[i] for row in rows if row[i] is not None]
if values:
analysis["summary"][col] = {
"min": min(values),
"max": max(values),
"avg": statistics.mean(values),
"median": statistics.median(values)
}
return analysis
def suggest_charts(self, analysis):
"""推荐可视化图表"""
suggestions = []
# 根据分析类型推荐图表
if "trend" in str(analysis).lower():
suggestions.append({
"type": "line",
"description": "折线图,适合展示趋势变化"
}
"""
数据分析 Agent
理解数据需求,执行分析,生成报告
"""
def __init__(self, llm, db_connection, chart_generator):
self.llm = llm
self.db = db_connection
self.chart_generator = chart_generator
def analyze(self, request):
"""
处理数据分析请求
"""
# 第一步:理解分析需求
intent = self.parse_intent(request)
# 第二步:生成查询
query = self.generate_query(intent)
# 第三步:执行查询
data = self.execute_query(query)
# 第四步:分析数据
analysis = self.perform_analysis(data, intent)
# 第五步:生成可视化建议
charts = self.suggest_charts(analysis)
# 第六步:生成报告
report = self.generate_report(analysis, charts)
return {
"intent": intent,
"query": query,
"data": data,
"analysis": analysis,
"charts": charts,
"report": report
}
def parse_intent(self, request):
"""解析用户意图"""
prompt = f"""
分析以下数据请求,提取关键信息:
请求:{request}
请提取:
1. 分析目标(比较、趋势、分布等)
2. 涉及的表和字段
3. 时间范围(如果有)
4. 筛选条件(如果有)
"""
return self.llm.generate(prompt)
def generate_query(self, intent):
"""生成 SQL 查询"""
prompt = f"""
基于以下意图生成 SQL 查询:
{intent}
数据库 Schema:
{self.get_schema()}
"""
return self.llm.generate(prompt)
def execute_query(self, query):
"""执行查询"""
cursor = self.db.cursor()
cursor.execute(query)
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
return {"columns": columns, "rows": rows}
def perform_analysis(self, data, intent):
"""执行数据分析"""
# 简化实现
import statistics
rows = data["rows"]
columns = data["columns"]
analysis = {
"row_count": len(rows),
"summary": {}
}
# 对数值列计算统计信息
for i, col in enumerate(columns):
if self.is_numeric(rows, i):
values = [row[i] for row in rows if row[i] is not None]
if values:
analysis["summary"][col] = {
"min": min(values),
"max": max(values),
"avg": statistics.mean(values),
"median": statistics.median(values)
}
return analysis
def suggest_charts(self, analysis):
"""推荐可视化图表"""
suggestions = []
# 根据分析类型推荐图表
if "trend" in str(analysis).lower():
suggestions.append({
"type": "line",
"description": "折线图,适合展示趋势变化"
}