Appearance
章节3：企业级应用开发
本章导学
AI Agent 技术在企业中的落地是当前最受关注的方向之一。本章将深入5个典型的企业级应用场景：知识库问答系统（RAG+Agent）、智能客服与坐席助手、文档处理与合同审核Agent、代码助手与研发效能Agent、数据分析师Agent。每个场景都会给出架构设计和核心代码实现，帮助你掌握从技术到产品的完整路径。
学习目标
- 掌握基于RAG+Agent的企业知识库问答系统架构
- 理解智能客服系统的自助+人工转接设计模式
- 能实现文档处理与合同审核的自动化流程
- 了解代码助手Agent的研发效能提升方案
- 学会构建自然语言驱动的数据分析Agent
3.1 企业知识库问答系统（RAG+Agent）
3.1.1 系统架构概述
企业知识库问答系统是企业AI落地的"第一选择"。核心架构采用RAG（Retrieval-Augmented Generation）+ Agent模式。
用户提问
│
▼
┌──────────────┐
│ 查询理解模块 │ <- Agent：分析意图、改写查询、分解复杂问题
└──────┬───────┘
│
▼
┌──────────────┐
│ 检索模块 │ <- RAG：向量检索 + 关键词检索 + 重排序
│ (ChromaDB) │
└──────┬───────┘
│
▼
┌──────────────┐
│ 增强生成模块 │ <- Agent：整合上下文、过滤冗余、生成回答
├──────────────┤
│ 引用溯源模块 │ <- 展示信息来源，提升可信度
└──────┬───────┘
│
▼
最终回答 + 引用3.1.2 核心代码实现
1）文档加载与向量化
python
# pip install langchain langchain-community chromadb sentence-transformers
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
import os
class KnowledgeBase:
"""企业知识库构建器"""
def __init__(self, persist_dir="./chroma_db"):
self.persist_dir = persist_dir
# 使用中文优化的嵌入模型
self.embeddings = HuggingFaceEmbeddings(
model_name="BAAI/bge-large-zh-v1.5",
model_kwargs={"device": "cpu"},
encode_kwargs={"normalize_embeddings": True}
)
def load_documents(self, data_dir: str) -> list:
"""加载企业文档"""
loader = DirectoryLoader(
data_dir,
glob="**/*.md",
loader_cls=TextLoader,
loader_kwargs={"encoding": "utf-8"}
)
documents = loader.load()
print(f"加载了 {len(documents)} 个文档")
return documents
def chunk_documents(self, documents: list) -> list:
"""文档分块"""
splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # 每块约500字
chunk_overlap=50, # 重叠50字保持上下文连贯
separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""]
)
chunks = splitter.split_documents(documents)
print(f"切分为 {len(chunks)} 个文本块")
return chunks
def build_vectorstore(self, chunks: list) -> Chroma:
"""构建向量数据库"""
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=self.embeddings,
persist_directory=self.persist_dir
)
vectorstore.persist()
print(f"向量数据库已保存到: {self.persist_dir}")
return vectorstore
def load_vectorstore(self) -> Chroma:
"""加载已有向量数据库"""
return Chroma(
persist_directory=self.persist_dir,
embedding_function=self.embeddings
)
# 使用示例
kb = KnowledgeBase()
# docs = kb.load_documents("./企业文档库")
# chunks = kb.chunk_documents(docs)
# vectorstore = kb.build_vectorstore(chunks)2）RAG检索增强生成
python
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
class RAGEngine:
"""RAG检索增强生成引擎"""
def __init__(self, vectorstore, model="gpt-4"):
self.vectorstore = vectorstore
self.llm = ChatOpenAI(model=model, temperature=0.1)
self.retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 5} # 检索Top-5相关片段
)
# 企业问答专用Prompt
self.prompt = PromptTemplate(
template="""你是一家企业的AI知识库助手。请基于以下提供的知识片段，
准确回答用户的问题。如果知识片段不足以回答，请明确说明。
知识片段：
{context}
用户问题：{question}
要求：
1. 只基于提供的知识片段回答
2. 如果知识不足，请说"根据现有知识库，我无法完整回答这个问题"
3. 请注明信息来源
回答：""",
input_variables=["context", "question"]
)
def query(self, question: str) -> dict:
"""查询知识库"""
chain = RetrievalQA.from_chain_type(
llm=self.llm,
chain_type="stuff",
retriever=self.retriever,
chain_type_kwargs={"prompt": self.prompt},
return_source_documents=True
)
result = chain.invoke({"query": question})
return {
"answer": result["result"],
"sources": [
doc.metadata.get("source", "未知来源")
for doc in result["source_documents"]
]
}
# 使用示例
# engine = RAGEngine(kb.load_vectorstore())
# response = engine.query("公司的年假政策是什么？")
# print(response["answer"])
# print("来源:", response["sources"])3）Agent增强版（带意图识别和多轮检索）
python
from openai import OpenAI
import json
class AgenticRAGEngine:
"""Agent增强版RAG引擎"""
def __init__(self, vectorstore, model="gpt-4"):
self.vectorstore = vectorstore
self.llm = OpenAI(model=model)
self.retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 3}
)
def analyze_intent(self, question: str) -> dict:
"""分析用户意图"""
response = self.llm.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": """分析用户问题的意图，输出JSON格式：
{
"intent": "knowledge_query|multi_step_query|ambiguous",
"sub_questions": ["子问题1", "子问题2"],
"requires_clarification": false,
"clarification_question": "需要澄清的问题（如有）"
}"""},
{"role": "user", "content": question}
],
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
def query(self, question: str) -> str:
"""带Agent增强的查询"""
# 1. 分析意图
intent = self.analyze_intent(question)
# 2. 如果是复杂问题，分解为子问题
if intent["intent"] == "multi_step_query":
answers = []
for sub_q in intent.get("sub_questions", [question]):
# 检索
docs = self.retriever.get_relevant_documents(sub_q)
context = "\n".join([d.page_content for d in docs])
# 生成
response = self.llm.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": f"基于上下文回答问题：\n{context}"},
{"role": "user", "content": sub_q}
]
)
answers.append(response.choices[0].message.content)
return "\n\n".join(answers)
# 3. 简单查询直接检索
docs = self.retriever.get_relevant_documents(question)
context = "\n".join([d.page_content for d in docs])
response = self.llm.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": f"你是企业知识库助手，基于上下文回答：\n{context}"},
{"role": "user", "content": question}
]
)
return response.choices[0].message.content3.1.3 企业级RAG架构关键优化
| 优化项 | 说明 | 实现方式 |
|---|---|---|
| 混合检索 | 向量检索+关键词检索互补 | Elasticsearch + ChromaDB |
| 重排序（Rerank） | 精排Top-K结果 | Cohere/dashscope Rerank API |
| HyDE | 假设性文档增强检索 | LLM先生成假设答案再检索 |
| 查询改写 | 将用户问句改写为更易检索的形式 | LLM Query Rewrite |
| 多轮对话 | 维护对话历史和上下文 | Session管理+历史摘要 |
3.2 智能客服与坐席助手
3.2.1 系统设计
企业级智能客服系统的核心是自助服务+人工转接的双轨模式。
用户提问
│
▼
┌──────────────┐
│ 智能分流模块 │ <- Agent：分析问题类型、紧急程度、客户等级
└──────┬───────┘
│
┌───┴────────┐
│ │
▼ ▼
┌────────┐ ┌──────────┐
│ FAQ自助 │ │ 人工坐席 │
│ (RAG+Bot)│ │ (辅助模式)│
└────────┘ └────┬─────┘
│
▼
┌──────────┐
│ 坐席助手 │ <- Agent：实时建议、知识推荐、话术生成
├──────────┤
│ 工单系统 │ <- 未解决问题自动生成工单
└──────────┘3.2.2 核心代码实现
python
from openai import OpenAI
import json
from enum import Enum
class IntentType(str, Enum):
FAQ = "faq" # 常见问题
TROUBLESHOOTING = "troubleshooting" # 故障排查
ORDER = "order_query" # 订单查询
COMPLAINT = "complaint" # 投诉
HUMAN = "need_human" # 需要人工
GENERAL = "general" # 一般咨询
class SmartCustomerService:
"""智能客服系统"""
def __init__(self, knowledge_base=None):
self.client = OpenAI()
self.knowledge_base = knowledge_base
def classify_intent(self, message: str, history: list = None) -> IntentType:
"""意图识别与分流"""
prompt = f"""用户消息：{message}
历史对话：{history or '无'}
请判断用户意图，只返回以下类型之一：
- faq: 常见问题咨询
- troubleshooting: 需要故障排查指导
- order_query: 订单或账户查询
- complaint: 投诉或不满
- need_human: 明确要求人工服务
- general: 一般性咨询
"""
response = self.client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}]
)
intent_str = response.choices[0].message.content.strip().lower()
for intent in IntentType:
if intent.value in intent_str:
return intent
return IntentType.GENERAL
def auto_reply(self, message: str, intent: IntentType) -> dict:
"""自助回复"""
# FAQ走RAG
if intent == IntentType.FAQ and self.knowledge_base:
context = self.knowledge_base.retrieve(message)
else:
context = ""
response = self.client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": f"""你是企业智能客服。
{context if context else '请根据通用知识回答。'}
回复要求：
1. 礼貌、专业、简洁
2. 如果无法解决，主动提供转人工选项
3. 故障排查请给出步骤
4. 投诉问题先道歉再解决"""},
{"role": "user", "content": message}
]
)
answer = response.choices[0].message.content
# 判断是否需要转人工
needs_handoff = any([
intent == IntentType.COMPLAINT,
intent == IntentType.HUMAN,
"转人工" in answer,
"无法解决" in answer
])
return {
"answer": answer,
"needs_handoff": needs_handoff,
"intent": intent.value
}
def agent_assistant(self, customer_msg: str, history: list) -> str:
"""坐席助手：为人工坐席提供实时建议"""
response = self.client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": """你是一名AI坐席助手，为客服人员提供实时支持。
基于对话历史和当前客户消息，输出：
1. 问题总结（一句话）
2. 建议回复（话术）
3. 知识库推荐（相关FAQ链接或知识点）
4. 客户情绪分析（积极/中性/消极/愤怒）"""},
{"role": "user", "content": f"历史对话：{history}\n当前消息：{customer_msg}"}
]
)
return response.choices[0].message.content
def handle(self, message: str, history: list = None):
"""完整处理流程"""
# 1. 意图识别
intent = self.classify_intent(message, history)
print(f"[智能分流] 意图：{intent.value}")
# 2. 自助回复
result = self.auto_reply(message, intent)
# 3. 如果需要转人工
if result["needs_handoff"]:
# 生成工单摘要
handoff_summary = f"问题类型：{intent.value}\n客户问题：{message}\n自助回复：{result['answer'][:100]}..."
print(f"[转人工] 已生成工单摘要")
result["handoff_summary"] = handoff_summary
return result
# 使用示例
cs = SmartCustomerService()
result = cs.handle("我想退掉昨天买的那个产品，质量太差了！")
print(f"回复：{result['answer']}")
print(f"需要转人工：{result['needs_handoff']}")3.2.3 关键指标
| 指标 | 说明 | 行业基准 |
|---|---|---|
| 自助解决率 | 用户不转人工即解决的问题比例 | >70% |
| 首响时间 | 用户发送到第一次回复的时间 | <5秒 |
| 解决率 | 用户问题被成功解决的比例 | >85% |
| 满意度 | 用户对服务的评分 | >4.0/5.0 |
| 转人工率 | 从机器人转到人工的比例 | <30% |
3.3 文档处理与合同审核Agent
3.3.1 系统架构
上传文档
│
▼
┌──────────────┐
│ 文档解析模块 │ <- PDF/Word/Excel多格式解析
├──────────────┤
│ 内容提取 │ <- OCR、表格提取、图片识别
└──────┬───────┘
│
▼
┌──────────────┐
│ 合同审核Agent │
│ - 关键条款提取 │
│ - 风险点识别 │
│ - 合规检查 │
│ - 建议修改 │
└──────┬───────┘
│
▼
┌──────────────┐
│ 审核报告生成 │ <- 结构化审核报告 + 标注版本
└──────────────┘3.3.2 核心代码实现
python
# pip install python-docx PyPDF2 openpyxl
import os
from openai import OpenAI
import json
class DocumentProcessor:
"""文档处理器：支持多格式解析"""
def extract_text(self, file_path: str) -> str:
"""提取文档文本内容"""
ext = os.path.splitext(file_path)[1].lower()
if ext == ".pdf":
return self._extract_pdf(file_path)
elif ext == ".docx":
return self._extract_docx(file_path)
elif ext == ".txt":
return self._read_txt(file_path)
else:
raise ValueError(f"不支持的文件格式: {ext}")
def _extract_pdf(self, file_path: str) -> str:
from PyPDF2 import PdfReader
reader = PdfReader(file_path)
text = ""
for page in reader.pages:
text += page.extract_text() + "\n"
return text
def _extract_docx(self, file_path: str) -> str:
from docx import Document
doc = Document(file_path)
text = "\n".join([p.text for p in doc.paragraphs])
return text
def _read_txt(self, file_path: str) -> str:
with open(file_path, "r", encoding="utf-8") as f:
return f.read()
class ContractReviewAgent:
"""合同审核Agent"""
CHECKLIST = [
"合同主体信息是否完整",
"合同金额和支付条款是否明确",
"违约责任条款是否合理",
"保密条款是否存在",
"争议解决方式是否明确",
"合同期限和终止条件",
"知识产权归属",
"不可抗力条款"
]
def __init__(self, model="gpt-4"):
self.client = OpenAI()
self.model = model
def review(self, contract_text: str) -> dict:
"""审核合同"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": f"""你是一名专业的企业法务顾问。
请审核以下合同内容，输出JSON格式的审核报告：
{{
"basic_info": {{
"contract_name": "合同名称",
"parties": ["甲方", "乙方"],
"contract_type": "合同类型"
}},
"checklist": [
{{
"item": "检查项",
"status": "pass|warn|fail",
"detail": "具体分析",
"suggestion": "修改建议"
}}
],
"risk_points": [
{{
"risk": "风险描述",
"severity": "high|medium|low",
"clause": "相关条款",
"recommendation": "应对建议"
}}
],
"overall_assessment": "总体评估",
"suggested_actions": ["建议操作1", "建议操作2"]
}}
请逐一检查以下要点：
{chr(10).join(f'- {item}' for item in CHECKLIST)}"""},
{"role": "user", "content": contract_text}
],
response_format={"type": "json_object"},
temperature=0.2
)
return json.loads(response.choices[0].message.content)
def generate_report(self, review_result: dict) -> str:
"""生成可读的审核报告"""
report = f"""
========================================
合同审核报告
========================================
### 基本信息
- 合同名称：{review_result['basic_info'].get('contract_name', '未识别')}
- 签约双方：{', '.join(review_result['basic_info'].get('parties', ['未识别']))}
- 合同类型：{review_result['basic_info'].get('contract_type', '未识别')}
### 逐项检查结果
"""
for item in review_result.get("checklist", []):
icon = {"pass": "✅", "warn": "⚠️", "fail": "❌"}
status_icon = icon.get(item["status"], "❓")
report += f"\n{status_icon} **{item['item']}**\n"
report += f" 分析：{item.get('detail', '')}\n"
if item.get("suggestion"):
report += f" 建议：{item.get('suggestion', '')}\n"
report += "\n### 风险点\n"
for risk in review_result.get("risk_points", []):
severity_icon = {"high": "🔴", "medium": "🟡", "low": "🟢"}
report += f"\n{severity_icon.get(risk['severity'], '⚪')} **{risk['risk']}** (严重程度: {risk['severity']})\n"
report += f" 相关条款：{risk.get('clause', '')}\n"
report += f" 应对建议：{risk.get('recommendation', '')}\n"
report += f"\n### 总体评估\n{review_result.get('overall_assessment', '')}\n"
report += "\n### 建议操作\n"
for action in review_result.get("suggested_actions", []):
report += f"- {action}\n"
return report
# 使用示例
# processor = DocumentProcessor()
# text = processor.extract_text("合同文件.docx")
# agent = ContractReviewAgent()
# result = agent.review(text)
# report = agent.generate_report(result)
# print(report)3.3.3 文档Agent的高级能力
| 能力 | 技术实现 | 应用场景 |
|---|---|---|
| 多文档对比 | 交叉比对+差异检测 | 合同版本对比 |
| 表格提取 | OCR+结构化提取 | 财务报表分析 |
| 关键信息抽取 | NER+LLM提取 | 发票/表单信息 |
| 合规检查 | 规则引擎+LLM | 行业合规审核 |
| 自动归档 | 元数据提取+分类 | 文档管理系统 |
3.4 代码助手与研发效能Agent
3.4.1 核心功能
企业研发效能Agent涵盖以下场景：
| 功能 | 描述 | 技术实现 |
|---|---|---|
| 代码生成 | 根据需求生成代码片段 | LLM + 上下文感知 |
| 代码审查 | 自动检查代码质量和安全问题 | 静态分析 + LLM Review |
| Bug修复 | 自动诊断和修复代码缺陷 | 错误日志分析 + 补丁生成 |
| 文档生成 | 从代码生成API文档和注释 | AST解析 + LLM |
| 架构建议 | 根据需求推荐技术方案 | 系统设计知识库 + LLM |
3.4.2 核心代码实现
python
from openai import OpenAI
import json
import ast
import subprocess
class CodeAssistantAgent:
"""代码助手Agent"""
def __init__(self, model="gpt-4"):
self.client = OpenAI()
self.model = model
def generate_code(self, requirements: str, language: str = "python") -> str:
"""根据需求生成代码"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": f"""你是一名高级{language}工程师。
请根据需求生成高质量代码。要求：
1. 包含完整的函数/类定义
2. 添加详细的文档字符串
3. 包含类型注解
4. 遵循PEP 8规范
5. 包含错误处理"""},
{"role": "user", "content": f"需求：{requirements}"}
]
)
return response.choices[0].message.content
def review_code(self, code: str) -> dict:
"""代码审查"""
# 先做静态分析
static_issues = self._static_analysis(code)
# 再用LLM做深度审查
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": """你是一名资深代码审查专家。请审查以下代码，输出JSON格式：
{
"overall_rating": "A/B/C/D",
"issues": [
{
"type": "bug|security|performance|style|maintainability",
"severity": "critical|major|minor",
"line": 行号,
"description": "问题描述",
"suggestion": "修改建议"
}
],
"strengths": ["代码优势1", "代码优势2"],
"summary": "总体评价"
}"""},
{"role": "user", "content": f"代码：\n```python\n{code}\n```"}
],
response_format={"type": "json_object"}
)
llm_review = json.loads(response.choices[0].message.content)
llm_review["static_issues"] = static_issues
return llm_review
def _static_analysis(self, code: str) -> list:
"""静态代码分析"""
issues = []
try:
tree = ast.parse(code)
# 检查缺少文档字符串
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.ClassDef)):
if not ast.get_docstring(node):
issues.append({
"type": "style",
"line": node.lineno,
"description": f"缺少文档字符串: {node.name}"
})
except SyntaxError as e:
issues.append({
"type": "bug",
"line": e.lineno or 0,
"description": f"语法错误: {e.msg}"
})
return issues
def fix_bug(self, code: str, error_message: str) -> str:
"""自动修复Bug"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一名代码调试专家。请修复以下代码中的Bug。"},
{"role": "user", "content": f"代码：\n{code}\n\n错误信息：{error_message}\n\n请输出修复后的完整代码："}
]
)
return response.choices[0].message.content
def generate_tests(self, code: str) -> str:
"""生成单元测试"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一名测试工程师。请为以下代码生成pytest单元测试，要求覆盖正常路径和边界情况。"},
{"role": "user", "content": f"代码：\n{code}"}
]
)
return response.choices[0].message.content
# 使用示例
agent = CodeAssistantAgent()
# 生成代码
code = agent.generate_code("实现一个LRU缓存类，支持get和put操作，容量限制为maxsize")
print(code)
# 审查代码
review = agent.review_code("""
def add(a,b):
return a+b
""")
print(json.dumps(review, ensure_ascii=False, indent=2))3.4.3 企业研发效能指标
| 指标 | 说明 | 提升效果 |
|---|---|---|
| 代码编写效率 | 单位时间完成的功能点 | 提升30-50% |
| Bug率 | 每千行代码缺陷数 | 降低20-40% |
| 代码审查周期 | PR从提交到合并的时间 | 缩短50-70% |
| 文档覆盖率 | 有文档的代码比例 | 提升至80%+ |
| 新人上手时间 | 新开发者达到生产效率的时间 | 缩短40-60% |
3.5 数据分析师Agent
3.5.1 系统设计
python
自然语言查询
│
▼
┌──────────────┐
│ NL2SQL模块 │ <- Agent: 自然语言转SQL查询
├──────────────┤
│ SQL执行引擎 │ <- 安全执行 + 结果缓存
├──────────────┤
│ 结果解释模块 │ <- Agent: 将数据结果转化为自然语言
├──────────────┤
│ 可视化建议 │ <- Agent: 推荐最佳可视化方式
└──────┬───────┘
│
▼
问答结果 + 图表3.5.2 核心代码实现
python
from openai import OpenAI
import json
import sqlite3
import pandas as pd
class DataAnalystAgent:
"""数据分析师Agent"""
def __init__(self, db_path: str, model="gpt-4"):
self.client = OpenAI()
self.model = model
self.db_path = db_path
self.schema = self._get_schema()
def _get_schema(self) -> str:
"""获取数据库结构"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = cursor.fetchall()
schema = []
for table in tables:
table_name = table[0]
cursor.execute(f"PRAGMA table_info({table_name})")
columns = cursor.fetchall()
col_desc = ", ".join([f"{col[1]} ({col[2]})" for col in columns])
schema.append(f"表 {table_name}: ({col_desc})")
conn.close()
return "\n".join(schema)
def nl2sql(self, question: str) -> str:
"""自然语言转SQL"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": f"""你是一个数据库查询专家。
数据库结构如下：
{self.schema}
请将用户的自然语言问题转换为SQL查询。
只输出SQL语句，不要加任何解释。"""},
{"role": "user", "content": f"问题：{question}"}
]
)
return response.choices[0].message.content.strip()
def execute_sql(self, sql: str) -> dict:
"""安全执行SQL查询"""
# 安全检查：只允许SELECT
if not sql.strip().upper().startswith("SELECT"):
return {"error": "只允许执行SELECT查询", "data": None}
try:
conn = sqlite3.connect(self.db_path)
df = pd.read_sql_query(sql, conn)
conn.close()
return {
"data": df.to_dict(orient="records"),
"columns": df.columns.tolist(),
"rows": len(df)
}
except Exception as e:
return {"error": str(e), "data": None}
def explain_result(self, question: str, sql: str, result: dict) -> str:
"""解释查询结果"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": """你是一名数据分析师。
请用通俗易懂的语言解释查询结果。包含：
1. 核心发现（关键数据点）
2. 数据解读（趋势、异常、洞察）
3. 业务建议（基于数据的行动建议）"""},
{"role": "user", "content": f"""
用户问题：{question}
SQL查询：{sql}
查询结果（共{result['rows']}条）：
{json.dumps(result['data'][:10], ensure_ascii=False, indent=2)}
""" if result['data'] else f"查询无结果或出错"}
]
)
return response.choices[0].message.content
def suggest_chart(self, result: dict) -> dict:
"""推荐可视化方式"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": """分析数据并推荐最佳可视化方式。
输出JSON格式：
{
"chart_type": "bar|line|pie|scatter|table",
"x_axis": "x轴字段",
"y_axis": "y轴字段",
"title": "图表标题",
"reason": "选择理由"
}"""},
{"role": "user", "content": f"数据列：{result['columns']}\n数据样例：{json.dumps(result['data'][:5], ensure_ascii=False)}"}
],
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
def ask(self, question: str) -> dict:
"""完整的数据分析流程"""
print(f"问题：{question}")
# 1. NL2SQL
sql = self.nl2sql(question)
print(f"SQL：{sql}")
# 2. 执行
result = self.execute_sql(sql)
if result.get("error"):
return {"error": result["error"]}
# 3. 结果解释
explanation = self.explain_result(question, sql, result)
# 4. 可视化建议
chart_suggestion = self.suggest_chart(result) if result["rows"] > 0 else None
return {
"sql": sql,
"data": result["data"][:20], # 限制返回条数
"explanation": explanation,
"chart": chart_suggestion
}
# ========= 快速开始：创建示例数据库 =========
def create_sample_db():
"""创建示例销售数据库"""
conn = sqlite3.connect("sales.db")
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS sales (
id INTEGER PRIMARY KEY,
product TEXT,
category TEXT,
amount REAL,
quantity INTEGER,
sale_date TEXT,
region TEXT
)
""")
sample_data = [
("笔记本电脑", "电子产品", 5999, 10, "2025-01-15", "华东"),
("手机", "电子产品", 3999, 25, "2025-01-20", "华南"),
("办公椅", "家具", 1299, 8, "2025-02-10", "华东"),
("显示器", "电子产品", 2499, 15, "2025-02-18", "华北"),
("键盘", "配件", 399, 30, "2025-03-05", "华南"),
]
cursor.executemany(
"INSERT INTO sales (product, category, amount, quantity, sale_date, region) VALUES (?, ?, ?, ?, ?, ?)",
sample_data
)
conn.commit()
conn.close()
print("示例数据库已创建：sales.db")
# 使用示例
# create_sample_db() # 首次运行取消注释
# agent = DataAnalystAgent("sales.db")
# result = agent.ask("各产品类别的总销售额是多少？")
# print(result["explanation"])3.5.3 安全与权限控制
| 安全措施 | 说明 | 实现方式 |
|---|---|---|
| 查询白名单 | 只允许SELECT操作 | SQL语法解析校验 |
| 行级权限 | 不同用户只能看特定数据 | WHERE条件自动注入 |
| 查询超时 | 防止慢查询 | SQL超时设置（<30s） |
| 结果脱敏 | 敏感字段自动隐藏 | 列级脱敏规则 |
| 审计日志 | 记录所有查询操作 | 日志系统 |
小结与练习
知识小结
- RAG+Agent知识库：文档向量化→混合检索→增强生成，是企业AI落地的首选方案
- 智能客服系统：意图识别→自助回复→转人工，双轨设计兼顾效率与体验
- 合同审核Agent：多格式解析→结构化审核→风险识别，提升法务效率
- 代码助手Agent：代码生成→审查→修复→测试，贯穿研发全流程
- 数据分析Agent：NL2SQL→执行→解释→可视化，降低数据使用门槛
练习
练习1：搭建企业知识库
选择你熟悉的技术文档（至少5篇），实现：
1. 文档加载和分块
2. 向量数据库构建（ChromaDB）
3. RAG问答接口
4. 添加引用溯源功能练习2：智能客服系统
实现一个客服Demo，包含：
1. 意图识别（至少4种意图）
2. FAQ自助回复（基于知识库）
3. 转人工逻辑
4. 坐席助手的建议生成练习3：合同审核工具
实现一个简单的合同审核工具：
1. 支持PDF/DOCX格式上传
2. 提取合同关键条款
3. 识别风险点（至少5类）
4. 生成审核报告练习4：数据分析Agent
连接一个数据库（SQLite/MySQL），实现：
1. 自然语言转SQL
2. 安全查询执行
3. 结果解释
4. 可视化建议
用3个不同的问题测试效果。下一章预告：章节4-生产级部署工程 - 从开发到生产环境的完整部署方案