前言
上周参加Google Cloud Next的预热直播，看到一个数据让我挺震撼的：截至2025年底，超过12万家企业在使用Gemini模型，付费席位超过800万。这个数字意味着什么？意味着企业级AI应用已经从”要不要用”变成了”怎么用好”的问题。
作为一个长期关注开发者工具的人，我一直想找机会系统地聊聊企业级AI工具的使用。之前写的很多教程都是面向个人开发者，这次换个角度，聊聊企业在生产环境中怎么用AI。
这篇文章会以Google Vertex AI为主要案例，讲解企业级AI应用的全流程：Agent构建、模型管理、数据安全、成本控制。不管你是企业的技术负责人，还是想在职业发展中了解企业AI应用的开发者，都能找到有用的内容。
为什么企业需要专门的AI平台？
先回答一个基础问题：个人开发者用ChatGPT挺好的，企业为什么要花钱买Vertex AI这样的平台？
这个问题其实很实际。总结下来，企业级AI平台解决的是四类核心问题：
第一是数据安全。企业的核心数据不能随便发给第三方API，但Gemini、GPT这些模型都需要云端处理。企业级平台提供私有部署和数据隔离，确保敏感信息不外泄。
第二是统一管理。公司里几十上百个AI应用，如果每个团队自己对接API，密钥管理、计费统计、权限控制都会乱套。统一平台可以集中管理所有AI能力。
第三是定制化需求。通用模型在垂直领域的表现往往不够好，企业需要用自己数据微调模型。企业级平台提供完整的模型微调和部署能力。
第四是合规审计。金融、医疗等行业对AI决策有严格的合规要求，需要完整的操作日志和审计追踪。
Vertex AI核心概念解析
在说具体操作之前，先把Vertex AI的几个核心概念讲清楚。理解这些，后面的实践会顺畅很多。
Vertex AI的整体架构
plaintext
┌─────────────────────────────────────────────────────────────┐
│ Vertex AI 平台 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────┐ │
│ │ Model │ │ Vertex AI │ │ Vertex AI │ │
│ │ Garden │ → │ Agent │ → │ Search │ │
│ │ (模型库) │ │ Builder │ │ (企业搜索) │ │
│ └─────────────┘ └─────────────┘ └─────────────────┘ │
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────┐ │
│ │ Tuning │ │ Workbench │ │ Feature │ │
│ │ (模型微调) │ │ (开发环境) │ │ Store │ │
│ └─────────────┘ └─────────────┘ └─────────────────┘ │
│ │
├─────────────────────────────────────────────────────────────┤
│ 底层基础设施 │
│ (TPU / GPU / Cloud Storage / BigQuery) │
└─────────────────────────────────────────────────────────────┘
- Model Garden：预置了大量模型，包括Gemini、Claude、Llama等，可以直接调用
- Agent Builder：无代码/低代码构建AI Agent的平台
- Vertex AI Search：企业级语义搜索
- Tuning：用企业数据微调模型
- Workbench：Jupyter风格的开发环境
- Feature Store：管理ML特征的统一存储
Vertex AI Agent的核心组件
python
# 官方Python SDK的核心概念映射
from vertexai import agent
from vertexai.language_models import TextGenerationModel
from vertexai.search import VertexSearch
# 1. 基础模型
base_model = TextGenerationModel.from_pretrained("gemini-2.0-flash")
# 2. Agent定义
my_agent = agent.Builder(
project="your-project-id",
location="us-central1",
# Agent的指令/角色定义
instruction="""你是一个客服助手，帮助用户解决账户问题。
- 优先使用公司的FAQ知识库回答问题
- 如果无法回答，收集用户问题并转人工
- 不要承诺超出服务范围的事情
""",
# 配备的工具
tools=[
# 语义搜索工具
VertexSearch(
data_store="customer-support-datastore",
max_results=5
),
# 预留其他工具接口
]
).build()
实战一：构建客服问答Agent
光说不练假把式，咱们直接上手构建一个客服问答Agent。这个场景很常见，很多企业都有这方面的需求。
第一步：准备知识库数据
企业客服Agent的效果，很大程度上取决于知识库的质量。我见过很多失败的案例，问题不在技术，而在于知识库内容太烂。
python
# 准备FAQ数据（JSON格式）
faq_data = [
{
"question": "如何重置密码？",
"answer": """重置密码有以下两种方式：
方式一：自助重置（推荐）
1. 点击登录页的"忘记密码"
2. 输入注册邮箱
3. 查收邮件，点击重置链接
4. 设置新密码（8-20位，需包含字母和数字）
方式二：联系客服
如果无法自助重置，请联系 support@company.com
注意：新密码设置后需30分钟才能生效。"""
},
{
"question": "如何取消订阅？",
"answer": """取消订阅的步骤：
1. 登录账号，进入"账户设置"
2. 点击"订阅管理"
3. 选择要取消的订阅套餐
4. 点击"取消订阅"并确认
重要提示：
- 取消后服务将继续至当前计费周期结束
- 已支付费用不予退还
- 取消后可随时重新订阅"""
},
# 更多FAQ...
]
# 将数据导入Vertex AI Search
from vertexai.resources.preview import rag
# 创建RAG语料库
rag_corpus = rag.create_corpus(
display_name="customer-support-corpus",
description="客服问答知识库"
)
# 导入文档
for item in faq_data:
rag.import_data(
corpus_name=rag_corpus.name,
paths=[
rag.RagFileData(
display_name=item["question"],
source_uri=f"data://faq/{item['question']}", # 实际场景需要真实存储
rag_file_content=item["answer"]
)
]
)
print(f"知识库创建完成，共导入 {len(faq_data)} 条FAQ")
第二步：构建Agent
python
# 构建客服Agent
from vertexai import agent
customer_service_agent = (
agent.Builder(
project="your-project-id",
location="us-central1"
)
.set_instruction("""
你是一家科技公司的智能客服助手，名叫"小助手"。
工作原则：
1. 首先在知识库中搜索相关答案，优先给出准确信息
2. 回答要简洁明了，避免过多专业术语
3. 如果知识库没有相关内容，坦诚告知用户
4. 收集用户信息时，说明收集目的
5. 遇到紧急问题（如账户被盗、支付异常），引导至人工客服
禁止行为：
- 不要承诺退款、赔偿等超出权限的事宜
- 不要透露用户隐私信息
- 不要给出法律、医疗等专业领域的建议
""")
.add_rag_corpus(rag_corpus.name)
.set_temperature(0.3) # 客服场景需要稳定性，降低创造性
.set_max_output_tokens(1024)
.build()
)
第三步：测试与调优
python
# 测试函数
def test_agent(agent, test_cases):
"""测试Agent的回答质量"""
results = []
for case in test_cases:
print(f"\n测试问题: {case['question']}")
response = agent.predict(input=case['question'])
answer = response.text
print(f"AI回答: {answer[:200]}...") # 截断显示
# 简单评估（实际项目需要更复杂的评估逻辑）
score = evaluate_answer(answer, case.get('criteria', ''))
results.append({
'question': case['question'],
'answer': answer,
'score': score
})
# 汇总报告
avg_score = sum(r['score'] for r in results) / len(results)
print(f"\n平均得分: {avg_score:.2f}/10")
return results
# 评估函数（简化版）
def evaluate_answer(answer, criteria):
"""评估回答质量"""
score = 10
# 检查是否为空
if not answer or len(answer) < 20:
score -= 3
# 检查是否表达了"不知道"
if any(kw in answer for kw in ['抱歉', '无法', '不知道', '无法回答']):
score -= 2
# 检查回答长度（太短或太长都不好）
if len(answer) > 2000:
score -= 1
return max(0, score)
# 执行测试
test_cases = [
{
'question': '密码忘了怎么办？',
'criteria': '应包含重置步骤'
},
{
'question': '我不想用了，能退钱吗？',
'criteria': '应引导至退款政策或人工客服'
},
{
'question': '你们公司是什么时候成立的？',
'criteria': '应坦诚表示不在知识库中'
}
]
results = test_agent(customer_service_agent, test_cases)
实战二：模型微调打造专属AI
通用模型在特定场景下表现不够好，这时候需要微调。Vertex AI提供了完整的微调能力。
什么时候需要微调？
微调不是万能的，在决定之前先问自己几个问题：
- 通用模型在这个场景的错误率是多少？
- 有多少标注数据可用？（通常需要1000+条）
- 这个场景会长期使用吗？
- 微调的成本能接受吗？
如果通用模型表现已经不错（比如95%以上的准确率），微调的收益可能不明显。但如果需要模型学习特定的格式、语气、专业术语，微调就很必要。
微调实战
python
# 第一步：准备微调数据
# 格式要求：每行一个JSON，包含input和output
training_data = [
{"input": "请分析一下Q3的销售数据", "output": "好的，我来为您分析Q3的销售数据...\n\n整体来看，Q3销售额较Q2增长15%，其中华东地区表现最佳，同比增长25%。\n\n主要增长驱动：\n1. 线上渠道增长30%\n2. 新产品线贡献20%增量\n\n建议关注：华南地区下滑5%，建议下周详细复盘。"},
{"input": "对比一下北京和上海的业绩", "output": "北京vs上海业绩对比：\n\n【北京】\n- 销售额：800万\n- 增长率：12%\n- 主力产品：A系列\n\n【上海】\n- 销售额：950万\n- 增长率：18%\n- 主力产品：B系列\n\n【结论】\n上海整体表现更优，但北京在客户留存率上领先。"},
# 更多训练数据...
]
# 保存为JSONL格式（每行一个JSON）
import json
with open("training_data.jsonl", "w", encoding="utf-8") as f:
for item in training_data:
f.write(json.dumps(item, ensure_ascii=False) + "\n")
# 第二步：上传数据
from google.cloud import storage
storage_client = storage.Client()
bucket = storage_client.bucket("your-training-data-bucket")
blob = bucket.blob("fine-tuning/training_data.jsonl")
blob.upload_from_filename("training_data.jsonl")
training_data_uri = f"gs://your-training-data-bucket/fine-tuning/training_data.jsonl"
# 第三步：创建微调任务
from vertexai.preview import tuning
tuning_job = tuning.TuningJob(
display_name="sales-report-generator-v1",
source_model="gemini-2.0-flash",
training_data=training_data_uri,
# 微调参数
train_steps=1000, # 训练步数，越多越精细但越贵
learning_rate_multiplier=1.0, # 学习率倍数
).start()
print(f"微调任务ID: {tuning_job.resource_name}")
print("微调进行中，预计需要30-60分钟...")
# 等待完成
tuning_job.wait()
print(f"微调完成！模型ID: {tuning_job.tuned_model_endpoint_name}")
验证微调效果
python
# 微调后对比测试
def compare_models(prompt, original_model, tuned_model):
"""对比原始模型和微调模型的效果"""
print(f"测试提示: {prompt}\n")
# 原始模型回答
original_response = original_model.predict(prompt)
print(f"【原始Gemini回答】\n{original_response.text[:500]}...\n")
# 微调模型回答
tuned_response = tuned_model.predict(prompt)
print(f"【微调后模型回答】\n{tuned_response.text[:500]}...")
return {
'original': original_response.text,
'tuned': tuned_response.text
}
# 运行对比
test_prompt = "请用表格形式总结本周各区域销售情况，包括销售额、增长率、环比变化"
result = compare_models(
test_prompt,
base_model,
tuned_model
)
实战三：企业级安全与权限管理
这是企业级应用的重头戏。前面讲的再花哨，如果安全没做好，都是白搭。
多租户数据隔离
python
from google.cloud import aiplatform
from google.auth import default
# 初始化AI平台
aiplatform.init(
project="your-enterprise-project",
location="us-central1",
# 启用数据 lineage 追踪
experiment="customer-ai-project"
)
# 创建租户隔离配置
class TenantIsolation:
"""确保不同租户之间的数据隔离"""
def __init__(self, tenant_id):
self.tenant_id = tenant_id
self._setup_permissions()
def _setup_permissions(self):
"""设置租户专属权限"""
# 获取当前认证信息
credentials, project = default()
# 租户只能访问自己前缀的存储桶
self.data_bucket = f"tenant-{self.tenant_id}-data"
self.model_bucket = f"tenant-{self.tenant_id}-models"
self.log_bucket = f"tenant-{self.tenant_id}-logs"
def get_allowed_paths(self):
"""获取当前租户允许访问的资源路径"""
return [
f"gs://{self.data_bucket}/**",
f"gs://{self.model_bucket}/**",
f"gs://{self.log_bucket}/**",
]
# 使用示例：为不同客户创建隔离环境
tenant_a = TenantIsolation("customer-a")
tenant_b = TenantIsolation("customer-b")
print(f"租户A数据路径: {tenant_a.get_allowed_paths()}")
print(f"租户B数据路径: {tenant_b.get_allowed_paths()}")
完整的审计日志
python
import logging
from datetime import datetime
from google.cloud import logging as cloud_logging
class AIAuditLogger:
"""AI平台审计日志"""
def __init__(self, project_id):
self.project_id = project_id
# 配置Cloud Logging
self.client = cloud_logging.Client(project=project_id)
self.logger = self.client.logger("ai-platform-audit")
# 敏感字段白名单
self.sensitive_fields = [
"password", "token", "secret", "api_key",
"ssn", "credit_card", "phone", "email"
]
def log_ai_interaction(
self,
user_id: str,
agent_id: str,
input_data: dict,
output_data: dict,
metadata: dict = None
):
"""记录AI交互日志"""
# 脱敏处理
safe_input = self._mask_sensitive(input_data)
safe_output = self._mask_sensitive(output_data)
log_entry = {
"timestamp": datetime.utcnow().isoformat(),
"event_type": "ai_interaction",
"user_id": user_id,
"agent_id": agent_id,
"input_preview": str(safe_input)[:500],
"output_preview": str(safe_output)[:500],
"metadata": metadata or {}
}
self.logger.log_struct(log_entry, severity="INFO")
def log_security_event(self, event_type: str, details: dict):
"""记录安全事件"""
log_entry = {
"timestamp": datetime.utcnow().isoformat(),
"event_type": "security_event",
"security_event_type": event_type,
"details": details
}
self.logger.log_struct(log_entry, severity="WARNING")
def _mask_sensitive(self, data: dict) -> dict:
"""脱敏处理"""
import copy
safe_data = copy.deepcopy(data)
def mask_recursive(obj):
if isinstance(obj, dict):
for key in obj:
if any(s in key.lower() for s in self.sensitive_fields):
obj[key] = "***REDACTED***"
else:
mask_recursive(obj[key])
elif isinstance(obj, list):
for item in obj:
mask_recursive(item)
return obj
return mask_recursive(safe_data)
# 使用审计日志
audit_logger = AIAuditLogger("your-project-id")
# 记录正常交互
audit_logger.log_ai_interaction(
user_id="user-123",
agent_id="customer-service-v1",
input_data={"query": "如何重置密码"},
output_data={"response": "请访问忘记密码页面..."},
metadata={"session_id": "abc123"}
)
# 记录可疑行为
audit_logger.log_security_event(
event_type="prompt_injection_attempt",
details={
"user_id": "user-456",
"suspicious_input": "ignore previous instructions",
"action_taken": "input_rejected"
}
)
成本控制与预算告警
企业最怕的是什么？账单打爆。AI API按token计费，如果不控制，分分钟烧光预算。
python
from google.cloud import billing
from datetime import datetime, timedelta
class AICostController:
"""AI成本控制器"""
def __init__(self, project_id, budget_limit_usd=1000):
self.project_id = project_id
self.budget_limit = budget_limit_usd
self.daily_limit = budget_limit_usd / 30 # 日均限额
self.month_start = datetime.utcnow().replace(day=1)
# 获取成本数据
self.billing_client = billing.CloudBillingClient()
def get_current_spend(self) -> dict:
"""获取当前账单"""
# 简化实现，实际应调用Billing API
return {
"month_to_date": 450.00,
"daily_average": 45.00,
"projected_month_end": 1350.00, # 预计月底账单
"currency": "USD"
}
def check_budget(self) -> tuple[bool, str]:
"""检查是否超预算"""
spend = self.get_current_spend()
remaining = self.budget_limit - spend["month_to_date"]
if spend["projected_month_end"] > self.budget_limit:
return False, f"⚠️ 预警：预计月底账单 ${spend['projected_month_end']:.2f} 将超出预算"
if remaining < self.daily_limit:
return False, f"⚠️ 剩余预算 ${remaining:.2f} 低于日均限额"
return True, f"✓ 预算正常，剩余 ${remaining:.2f}"
def estimate_request_cost(self, input_tokens: int, output_tokens: int) -> float:
"""估算单次请求成本（以Gemini 2.0 Flash为例）"""
# 2026年参考价格
input_cost_per_m = 0.05 # 每百万token $0.05
output_cost_per_m = 0.15 # 每百万token $0.15
input_cost = (input_tokens / 1_000_000) * input_cost_per_m
output_cost = (output_tokens / 1_000_000) * output_cost_per_m
return input_cost + output_cost
# 集成到Agent
class CostAwareAgent:
"""带成本感知的Agent"""
def __init__(self, agent, cost_controller):
self.agent = agent
self.cost_controller = cost_controller
def predict(self, prompt):
# 预测输入token数（简单估算）
input_tokens = len(prompt) // 4 # 粗略估算
# 估算成本
estimated = self.cost_controller.estimate_request_cost(
input_tokens,
output_tokens=500 # 假设输出500 tokens
)
# 成本太高则拒绝
if estimated > 0.50: # 超过$0.5的单次请求需要审核
return {"error": "请求过大，请拆分问题", "estimated_cost": estimated}
# 检查预算
can_proceed, msg = self.cost_controller.check_budget()
if not can_proceed:
return {"error": msg}
# 执行请求
return self.agent.predict(prompt)
# 使用
cost_controller = AICostController("your-project", budget_limit_usd=2000)
agent = CostAwareAgent(customer_service_agent, cost_controller)
实战四：API封装与第三方集成
企业内部的AI能力，最终要开放给其他系统使用。最常见的方式是通过API封装。
python
from flask import Flask, request, jsonify
from functools import wraps
import time
app = Flask(__name__)
# 简单的Token认证
VALID_TOKENS = {
"app-internal-token": {"app": "internal-dashboard", "tier": "unlimited"},
"partner-api-token": {"app": "partner-system", "tier": "standard"}
}
def require_auth(f):
"""API认证装饰器"""
@wraps(f)
def decorated(*args, **kwargs):
token = request.headers.get("Authorization", "").replace("Bearer ", "")
if token not in VALID_TOKENS:
return jsonify({"error": "无效的访问令牌"}), 401
request.app_info = VALID_TOKENS[token]
return f(*args, **kwargs)
return decorated
# API路由
@app.route("/api/v1/ai/chat", methods=["POST"])
@require_auth
def chat():
"""对话接口"""
data = request.json
# 参数验证
if not data.get("message"):
return jsonify({"error": "message字段不能为空"}), 400
# 调用Agent
response = customer_service_agent.predict(data["message"])
return jsonify({
"success": True,
"response": response.text,
"model": "gemini-2.0-flash-tuned",
"tokens_used": {
"input": response.usage_metadata.prompt_token_count,
"output": response.usage_metadata.candidates_token_count
}
})
@app.route("/api/v1/ai/batch", methods=["POST"])
@require_auth
def batch_process():
"""批量处理接口"""
data = request.json
# 限制批量大小
messages = data.get("messages", [])
if len(messages) > 50:
return jsonify({"error": "单次批量请求最多50条"}), 400
# 限流：根据tier限制QPS
tier = request.app_info["tier"]
rate_limit = {"unlimited": 100, "standard": 10}[tier]
results = []
for msg in messages:
# 实际应该并发处理，这里简化
resp = customer_service_agent.predict(msg)
results.append({
"input": msg,
"output": resp.text
})
return jsonify({
"success": True,
"processed": len(results),
"results": results
})
# 健康检查
@app.route("/health", methods=["GET"])
def health():
return jsonify({"status": "healthy", "version": "1.0.0"})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8080)
总结与建议
聊了这么多企业级AI工具，最后总结几点我的看法：
关于技术选型：Vertex AI确实是个强大的平台，但不是唯一选择。AWS Bedrock、Azure OpenAI Service各有优势。选型时要考虑现有技术栈、数据合规要求、成本预算等因素。
关于落地建议：
- 先从简单场景试点，不要一上来就搞大项目
- 知识库质量比模型能力更重要
- 监控和成本控制要从第一天就做好
- 安全合规不能事后补救
关于团队能力：企业级AI应用需要复合型人才——既懂AI技术，又了解业务流程。建议组建专门的AI中台团队，负责能力建设和赋能，而不是每个业务团队自己搞。
关于未来趋势：Google Cloud Next ’26透露的方向很有意思——Agentic AI会成为主流。AI不再只是回答问题，而是要能自主规划、跨系统执行复杂任务。这个趋势值得持续关注。
技术发展很快，但企业的核心诉求不会变：降本增效、控制风险。希望这篇文章能帮你在AI落地的路上少走一些弯路。
发表回复