【珍藏必备】AI Agent开发实战：Google白皮书核心技术解析与架构设计全攻略
·
文章基于Google《Agents》白皮书，系统解析了AI Agent的工程化实践与技术框架。详细介绍了Agent的三层核心架构（模型层、工具层、编排层），分析了与传统LLM应用的本质区别，并提供了从架构设计到生产部署的完整技术路径，包括性能优化、成本控制、质量保证等关键实践，以及客服、代码助手、数据分析等实际应用场景案例。
前言
Google发布的《Agents》白皮书，为AI Agent的工程化实践提供了系统性的技术框架。作为一线开发者，我们需要的不是概念解释，而是可操作的技术方案。本文基于白皮书内容，结合实际开发经验，为Agent应用开发者提供从架构设计到生产部署的完整技术路径。
如果你正在或计划开发Agent应用，这篇文章将帮你避开常见的技术陷阱，选择合适的架构模式。
Agent的核心架构分为三个关键部分：模型、工具和编排层。
1. 模型层（Model）
这是Agent的“大脑”，负责核心决策。它可以是单个模型，也可以是多个模型的组合，例如使用一个模型负责规划，另一个模型负责执行，还有一个模型负责评估，以达到更好的性能。为了让模型更好地适应Agent任务，可以通过提供示例来对其进行微调，以展示其能力和工具使用方式。
2. 工具层（Tools）
工具是Agent与外部世界交互的“手脚”。它们弥补了模型无法直接感知和影响现实世界的局限性。白皮书详细介绍了三种核心工具类型：
- • Extensions（扩展）：它以标准化的方式连接Agent和外部API，让Agent能够无缝地执行API调用。Agent会根据用户查询、可用扩展和历史上下文来动态选择最合适的API。
- • Functions（函数）：这种模式将API的执行逻辑从Agent端转移到了客户端。模型仅输出函数名称和参数，而不进行实际的API调用。这种方式让开发者对数据流和执行流程有了更精细的控制，特别适用于对安全、认证或时序有严格要求的场景。
- • Data Stores（数据存储）：通过向量数据库实现检索增强生成（RAG），为Agent提供了访问动态、实时知识的能力。它通过将用户查询、文档等转换为向量嵌入（embeddings），然后进行相似度匹配，从而检索相关信息并提供给模型，使其能够生成基于事实的回答。
3. 编排层（Orchestration Layer）
这是Agent的“思维过程”，它定义了Agent如何进行推理、规划和决策。它是一个循环过程，接收信息、进行内部推理，并基于推理结果采取下一步行动或决策。
白皮书提到了几种主流的推理框架：
- • ReAct（Reasoning and Action）：这是一种提示工程框架，它让模型能通过“思考”和“行动”的循环来解决问题。Agent先进行内部思考（Thought），然后采取行动（Action），并从行动中获得观察结果（Observation），最后根据观察结果进行下一轮的思考和行动，直至得出最终答案。
- • Chain-of-Thought（CoT）：通过中间步骤来引导模型的推理过程，让复杂问题分解为更简单的子问题。
- • Tree-of-Thoughts（ToT）：这是一种更高级的框架，它允许模型探索多条“思维链”，适用于需要探索和战略性预判的任务。
Agent vs 传统LLM应用：核心差异与技术选型
技术架构的本质区别
传统LLM应用本质上是"问答系统"：用户提问→模型推理→返回结果。而Agent是"任务执行系统"：接收目标→制定计划→调用工具→执行任务→反馈调整。
这种本质上的差异，导致了两种应用在技术架构上的巨大分歧：
| 维度 | 传统LLM应用 | AI Agent |
|---|---|---|
| 知识范围 | 局限于训练数据 | 通过工具访问实时信息 |
| 交互模式 | 单次问答，无状态 | 多轮对话，有状态管理 |
| 执行能力 | 仅文本生成 | 可调用外部API、函数、数据库 |
| 推理架构 | 依赖提示工程 | 内置认知架构，多步推理 |
简而言之，Agent的能力是模型的超集。它在模型之上，构建了一套完整的感知-规划-执行-反馈循环。
这种差异体现在几个关键技术点：
1. 状态管理
# 传统LLM应用
defprocess_query(user_input):
response = llm.generate(user_input)
return response
# Agent应用
classAgentSession:
def__init__(self):
self.conversation_history = []
self.task_state = {}
self.available_tools = []
self.execution_plan = []
defprocess_task(self, user_goal):
# 维护会话状态，支持多轮交互
pass
2. 工具调用机制
传统应用中，工具调用通常是硬编码的条件判断。Agent需要智能决策何时使用哪个工具：
# 传统方式：硬编码工具调用
if"天气"in user_input:
result = weather_api.get_weather()
# Agent方式：智能工具选择
defselect_tools(task_context, available_tools):
# 基于任务上下文动态选择工具
tool_selection_prompt = f"""
Task: {task_context}
Available tools: {[tool.description for tool in available_tools]}
Select the most appropriate tools and explain why.
"""
return llm.analyze_and_select(tool_selection_prompt)
3. 推理链路
Agent具备多步推理能力，这是架构设计的关键：
classReActAgent:
defsolve_task(self, task):
thought = self.think(task) # 思考阶段
action = self.plan_action(thought) # 规划行动
observation = self.execute_action(action) # 执行并观察
ifself.is_task_complete(observation):
returnself.generate_final_response(observation)
else:
# 继续下一轮推理
returnself.solve_task(self.update_context(task, observation))
选型建议
选择传统LLM应用的场景：
- • 任务边界明确，不需要多步操作
- • 主要处理文本生成、分析、翻译等单一任务
- • 对实时性要求较高，不能接受多轮调用延迟
选择Agent架构的场景：
- • 需要与外部系统集成（数据库、API、文件系统）
- • 任务复杂，需要多步骤完成
- • 用户需求模糊，需要Agent主动澄清和规划
- • 需要处理异常情况和动态调整策略
Agent核心架构深度解析
1. 模型层设计策略
模型选择直接影响Agent性能。根据白皮书分析和实践经验：
单模型 vs 多模型架构
# 单模型架构：简单但可能存在能力瓶颈
classSingleModelAgent:
def__init__(self, model_name="gpt-4"):
self.llm = load_model(model_name)
defprocess(self, task):
returnself.llm.generate(task)
# 多模型架构：复杂但性能更优
classMultiModelAgent:
def__init__(self):
self.planner_model = load_model("gpt-4") # 规划能力强
self.executor_model = load_model("claude-3.5") # 代码执行好
self.critic_model = load_model("gemini-pro") # 结果评估
defprocess(self, task):
plan = self.planner_model.generate_plan(task)
result = self.executor_model.execute(plan)
evaluation = self.critic_model.evaluate(result)
returnself.integrate_results(result, evaluation)
模型性能优化要点：
-
- 提示词工程：为Agent设计结构化提示模板
AGENT_PROMPT_TEMPLATE = """
You are an AI agent designed to complete complex tasks.
AVAILABLE TOOLS:
{tools_description}
TASK: {user_task}
CONTEXT: {conversation_history}
INSTRUCTIONS:
1. Think step-by-step about how to complete this task
2. Select appropriate tools from the available options
3. Execute actions and observe results
4. If the task is not complete, continue with the next step
5. Provide a final summary when the task is done
Begin your reasoning:
"""
-
- 上下文窗口管理：
classContextManager:
def__init__(self, max_tokens=8192):
self.max_tokens = max_tokens
self.conversation_history = []
defadd_interaction(self, user_input, agent_response):
self.conversation_history.append({
'user': user_input,
'agent': agent_response,
'timestamp': time.time()
})
self._truncate_if_needed()
def_truncate_if_needed(self):
# 基于token数量和重要性截断历史记录
whileself._count_tokens() > self.max_tokens:
self.conversation_history.pop(0)
2. 工具系统实战设计
白皮书提到三种工具类型，实际开发中需要根据具体需求选择：
Extensions：API调用的标准化封装
适用场景：需要集成多个第三方API，希望Agent能智能选择调用。
classWeatherExtension:
def__init__(self, api_key):
self.api_key = api_key
self.description = "获取指定城市的天气信息"
self.parameters = {
"city": "string, required, 城市名称",
"days": "int, optional, 预报天数，默认1天"
}
defexecute(self, city, days=1):
try:
response = requests.get(
f"https://api.weather.com/v1/forecast",
params={"key": self.api_key, "q": city, "days": days}
)
return response.json()
except Exception as e:
return {"error": str(e)}
classExtensionManager:
def__init__(self):
self.extensions = {}
defregister(self, name, extension):
self.extensions[name] = extension
defget_tool_descriptions(self):
descriptions = []
for name, ext inself.extensions.items():
descriptions.append({
'name': name,
'description': ext.description,
'parameters': ext.parameters
})
return descriptions
defexecute_tool(self, tool_name, **kwargs):
if tool_name inself.extensions:
returnself.extensions[tool_name].execute(**kwargs)
else:
return {"error": f"Tool {tool_name} not found"}
Functions：客户端执行的安全调用
适用场景：涉及敏感操作、需要用户授权、或需要在特定环境执行。
classDatabaseFunction:
@staticmethod
defget_schema():
return {
"name": "query_database",
"description": "查询数据库并返回结果",
"parameters": {
"type": "object",
"properties": {
"sql": {"type": "string", "description": "SQL查询语句"},
"limit": {"type": "integer", "description": "返回结果数量限制"}
},
"required": ["sql"]
}
}
@staticmethod
defexecute(sql, limit=100):
# 客户端执行，可以进行安全检查
ifnot DatabaseFunction.is_safe_query(sql):
return {"error": "Unsafe query detected"}
# 执行查询逻辑
# ...
pass
@staticmethod
defis_safe_query(sql):
# 实现SQL安全检查逻辑
dangerous_keywords = ['DROP', 'DELETE', 'UPDATE', 'INSERT']
returnnotany(keyword in sql.upper() for keyword in dangerous_keywords)
# Agent调用示例
classFunctionCallingAgent:
def__init__(self):
self.available_functions = [DatabaseFunction.get_schema()]
defprocess_query(self, user_input):
# Agent生成函数调用
function_call = self.llm.generate_function_call(
user_input,
self.available_functions
)
if function_call['name'] == 'query_database':
# 返回函数调用给客户端执行
return {
"type": "function_call",
"function": function_call['name'],
"parameters": function_call['parameters']
}
return {"type": "text_response", "content": "..."}
Data Stores：RAG系统的工程化实现
这是最复杂也是最重要的工具类型。实际开发中需要考虑：
import faiss
import numpy as np
from sentence_transformers import SentenceTransformer
classProductionRAGSystem:
def__init__(self, embedding_model_name="sentence-transformers/all-MiniLM-L6-v2"):
self.embedding_model = SentenceTransformer(embedding_model_name)
self.vector_index = None
self.document_store = []
self.chunk_size = 512
self.overlap_size = 50
defadd_documents(self, documents):
"""添加文档到向量数据库"""
chunks = []
for doc in documents:
doc_chunks = self._chunk_document(doc)
chunks.extend(doc_chunks)
# 生成embedding
embeddings = self.embedding_model.encode(chunks)
# 构建FAISS索引
ifself.vector_index isNone:
dimension = embeddings.shape[1]
self.vector_index = faiss.IndexFlatIP(dimension)
# 标准化向量
faiss.normalize_L2(embeddings)
self.vector_index.add(embeddings.astype('float32'))
self.document_store.extend(chunks)
def_chunk_document(self, document):
"""文档分块策略"""
words = document.split()
chunks = []
for i inrange(0, len(words), self.chunk_size - self.overlap_size):
chunk_words = words[i:i + self.chunk_size]
chunk_text = ' '.join(chunk_words)
chunks.append(chunk_text)
return chunks
defsearch(self, query, top_k=5):
"""检索相关文档"""
ifself.vector_index isNone:
return []
query_embedding = self.embedding_model.encode([query])
faiss.normalize_L2(query_embedding)
scores, indices = self.vector_index.search(query_embedding.astype('float32'), top_k)
results = []
for score, idx inzip(scores[0], indices[0]):
if idx < len(self.document_store):
results.append({
'content': self.document_store[idx],
'score': float(score)
})
return results
defgenerate_answer(self, query, context_docs):
"""基于检索结果生成回答"""
context = '\n\n'.join([doc['content'] for doc in context_docs])
prompt = f"""
Based on the following context, answer the user's question.
Context:
{context}
Question: {query}
Answer:
"""
returnself.llm.generate(prompt)
# 使用示例
classRAGAgent:
def__init__(self):
self.rag_system = ProductionRAGSystem()
self.llm = load_model("gpt-4")
defadd_knowledge(self, documents):
self.rag_system.add_documents(documents)
defanswer_query(self, query):
# 检索相关文档
relevant_docs = self.rag_system.search(query, top_k=3)
# 生成回答
if relevant_docs:
returnself.rag_system.generate_answer(query, relevant_docs)
else:
return"抱歉，我没有找到相关信息来回答您的问题。"
这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】
3. 编排层：Agent的"大脑"设计
编排层是Agent的核心，决定了任务执行的智能程度。
ReAct框架实现
classReActAgent:
def__init__(self, llm, tools):
self.llm = llm
self.tools = {tool.name: tool for tool in tools}
self.max_iterations = 10
defsolve_task(self, task):
context = f"Task: {task}\n\n"
for iteration inrange(self.max_iterations):
# Thought: 分析当前情况
thought = self._generate_thought(context)
context += f"Thought {iteration + 1}: {thought}\n"
# Action: 决定下一步行动
action = self._generate_action(context)
context += f"Action {iteration + 1}: {action}\n"
# 解析行动
if action.startswith("Final Answer:"):
return action[len("Final Answer:"):].strip()
# 执行工具调用
try:
tool_name, tool_input = self._parse_action(action)
observation = self.tools[tool_name].execute(tool_input)
context += f"Observation {iteration + 1}: {observation}\n\n"
except Exception as e:
context += f"Observation {iteration + 1}: Error - {str(e)}\n\n"
return"任务执行超时，请简化任务或检查工具配置。"
def_generate_thought(self, context):
prompt = f"""
{context}
分析当前情况，思考下一步应该怎么做。
Thought:"""
response = self.llm.generate(prompt, max_tokens=200)
return response.strip()
def_generate_action(self, context):
tool_descriptions = '\n'.join([
f"- {name}: {tool.description}"
for name, tool inself.tools.items()
])
prompt = f"""
{context}
可用工具：
{tool_descriptions}
基于以上思考，选择下一步行动。格式：
- 使用工具：工具名称[输入参数]
- 结束任务：Final Answer: [最终答案]
Action:"""
response = self.llm.generate(prompt, max_tokens=100)
return response.strip()
def_parse_action(self, action):
# 解析 "工具名称[输入参数]" 格式
if'['in action and']'in action:
tool_name = action.split('[')[0].strip()
tool_input = action.split('[')[1].split(']')[0]
return tool_name, tool_input
else:
raise ValueError(f"无法解析行动格式: {action}")
改进版：带有错误恢复的Agent
classRobustAgent:
def__init__(self, llm, tools):
self.llm = llm
self.tools = tools
self.error_recovery_attempts = 3
defsolve_task(self, task):
try:
returnself._solve_with_recovery(task)
except Exception as e:
returnf"任务执行失败: {str(e)}"
def_solve_with_recovery(self, task):
context = f"Task: {task}\n\n"
errors = []
for iteration inrange(10):
try:
# 正常执行逻辑
result = self._execute_iteration(context)
if result.get('finished'):
return result['answer']
context = result['updated_context']
except Exception as e:
errors.append(str(e))
# 错误恢复策略
iflen(errors) <= self.error_recovery_attempts:
recovery_prompt = f"""
执行过程中遇到错误：{str(e)}
历史错误：{'; '.join(errors)}
请分析错误原因并调整策略。继续执行任务：{task}
"""
context += f"Error Recovery: {recovery_prompt}\n"
continue
else:
raise e
return"任务执行超时"
def_execute_iteration(self, context):
# 具体的执行逻辑
pass
性能优化实践
1. 推理延迟优化
Agent应用的最大问题是多轮调用导致的延迟。优化策略：
import asyncio
from concurrent.futures import ThreadPoolExecutor
classOptimizedAgent:
def__init__(self):
self.llm_pool = ThreadPoolExecutor(max_workers=3)
self.tool_cache = {} # 工具调用结果缓存
asyncdefparallel_tool_calls(self, tool_calls):
"""并行执行多个工具调用"""
tasks = []
for tool_call in tool_calls:
ifself._can_cache(tool_call):
cache_key = self._get_cache_key(tool_call)
if cache_key inself.tool_cache:
continue
task = asyncio.create_task(self._execute_tool_async(tool_call))
tasks.append(task)
results = await asyncio.gather(*tasks)
return results
def_can_cache(self, tool_call):
# 判断工具调用是否可以缓存
cacheable_tools = ['weather', 'static_data_query']
return tool_call['name'] in cacheable_tools
asyncdef_execute_tool_async(self, tool_call):
loop = asyncio.get_event_loop()
returnawait loop.run_in_executor(
self.llm_pool,
self._execute_tool_sync,
tool_call
)
2. 成本控制策略
classCostOptimizedAgent:
def__init__(self):
self.cost_tracker = {
'input_tokens': 0,
'output_tokens': 0,
'tool_calls': 0
}
self.cost_limits = {
'max_tokens_per_task': 10000,
'max_tool_calls_per_task': 20
}
defprocess_with_budget(self, task):
ifself._check_budget():
returnself._process_task(task)
else:
return"任务超出预算限制，请简化任务或增加预算。"
def_check_budget(self):
return (
self.cost_tracker['input_tokens'] + self.cost_tracker['output_tokens']
< self.cost_limits['max_tokens_per_task']
and
self.cost_tracker['tool_calls'] < self.cost_limits['max_tool_calls_per_task']
)
def_track_usage(self, input_tokens, output_tokens, tool_calls=0):
self.cost_tracker['input_tokens'] += input_tokens
self.cost_tracker['output_tokens'] += output_tokens
self.cost_tracker['tool_calls'] += tool_calls
3. 质量保证机制
classQualityAssuredAgent:
def__init__(self, primary_llm, validator_llm):
self.primary_llm = primary_llm
self.validator_llm = validator_llm
defsolve_task_with_validation(self, task):
# 主Agent执行任务
primary_result = self.primary_agent.solve_task(task)
# 验证器检查结果
validation_prompt = f"""
任务：{task}
执行结果：{primary_result}
请评估这个结果是否：
1. 正确回答了问题
2. 逻辑清晰
3. 没有明显错误
如果有问题，请指出具体问题。
评估结果：
"""
validation = self.validator_llm.generate(validation_prompt)
if"有问题"in validation or"错误"in validation:
# 重新执行或修正
correction_prompt = f"""
原始任务：{task}
初始结果：{primary_result}
发现的问题：{validation}
请基于问题反馈，重新执行任务或修正结果。
"""
corrected_result = self.primary_llm.generate(correction_prompt)
return corrected_result
return primary_result
生产环境部署指南
1. LangChain快速原型
适合MVP开发和概念验证：
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain_community.tools import DuckDuckGoSearchRun
# 快速搭建Agent
defcreate_production_agent():
# 初始化模型
llm = ChatOpenAI(model="gpt-4-1106-preview", temperature=0)
# 定义工具
search = DuckDuckGoSearchRun()
tools = [
Tool(
name="Search",
func=search.run,
description="搜索最新信息"
),
Tool(
name="Calculator",
func=lambda x: eval(x), # 生产环境需要安全的计算器
description="执行数学计算"
)
]
# 创建Agent
agent = create_openai_tools_agent(llm, tools, prompt_template)
return AgentExecutor(agent=agent, tools=tools, verbose=True)
# 使用示例
agent = create_production_agent()
result = agent.invoke({"input": "帮我查找最新的AI技术趋势"})
2. 自定义生产框架
对于复杂业务场景，推荐自建框架：
classProductionAgentFramework:
def__init__(self, config):
self.config = config
self.llm = self._init_llm()
self.tools = self._init_tools()
self.memory = self._init_memory()
self.monitor = self._init_monitoring()
def_init_llm(self):
# 根据配置初始化模型
model_config = self.config['model']
if model_config['provider'] == 'openai':
return ChatOpenAI(**model_config['params'])
elif model_config['provider'] == 'anthropic':
return ChatAnthropic(**model_config['params'])
# ... 其他模型
def_init_tools(self):
tools = []
for tool_config inself.config['tools']:
tool_class = self._get_tool_class(tool_config['type'])
tool = tool_class(**tool_config['params'])
tools.append(tool)
return tools
def_init_memory(self):
# 初始化记忆系统
ifself.config['memory']['type'] == 'redis':
return RedisMemory(**self.config['memory']['params'])
elifself.config['memory']['type'] == 'postgresql':
return PostgreSQLMemory(**self.config['memory']['params'])
else:
return InMemoryMemory()
def_init_monitoring(self):
# 初始化监控系统
return AgentMonitor(
metrics_backend=self.config['monitoring']['backend'],
alert_thresholds=self.config['monitoring']['thresholds']
)
defprocess_request(self, user_id, task):
withself.monitor.track_execution():
try:
# 加载用户会话
session = self.memory.get_session(user_id)
# 执行任务
result = self._execute_task(task, session)
# 保存会话状态
self.memory.save_session(user_id, session)
# 记录成功指标
self.monitor.record_success(task, result)
return result
except Exception as e:
# 记录错误
self.monitor.record_error(task, e)
raise e
3. 监控和运维
import time
import logging
from dataclasses import dataclass
from typing importDict, Any
@dataclass
classExecutionMetrics:
task_id: str
start_time: float
end_time: float
token_usage: Dict[str, int]
tool_calls: int
success: bool
error_message: str = None
classAgentMonitor:
def__init__(self):
self.metrics_store = []
self.alert_thresholds = {
'max_execution_time': 30.0, # 秒
'max_token_usage': 5000,
'error_rate_threshold': 0.1# 10%
}
deftrack_execution(self):
return ExecutionTracker(self)
defanalyze_performance(self, time_window_hours=24):
recent_metrics = self._get_recent_metrics(time_window_hours)
ifnot recent_metrics:
return"没有足够的数据进行分析"
# 计算性能指标
avg_execution_time = sum(m.end_time - m.start_time for m in recent_metrics) / len(recent_metrics)
success_rate = sum(1for m in recent_metrics if m.success) / len(recent_metrics)
avg_token_usage = sum(m.token_usage.get('total', 0) for m in recent_metrics) / len(recent_metrics)
# 生成报告
report = f"""
Agent性能报告 (最近{time_window_hours}小时):
- 平均执行时间: {avg_execution_time:.2f}秒
- 成功率: {success_rate:.2%}
- 平均Token使用量: {avg_token_usage:.0f}
- 处理任务数: {len(recent_metrics)}
"""
# 检查告警
if avg_execution_time > self.alert_thresholds['max_execution_time']:
report += f"\n⚠️ 执行时间超过阈值 ({self.alert_thresholds['max_execution_time']}s)"
if success_rate < (1 - self.alert_thresholds['error_rate_threshold']):
report += f"\n⚠️ 错误率过高 (>{self.alert_thresholds['error_rate_threshold']:.1%})"
return report
classExecutionTracker:
def__init__(self, monitor):
self.monitor = monitor
self.start_time = None
self.metrics = None
def__enter__(self):
self.start_time = time.time()
returnself
def__exit__(self, exc_type, exc_val, exc_tb):
end_time = time.time()
self.metrics = ExecutionMetrics(
task_id=str(time.time()),
start_time=self.start_time,
end_time=end_time,
token_usage={'total': 0}, # 需要从实际执行中获取
tool_calls=0,
success=exc_type isNone,
error_message=str(exc_val) if exc_val elseNone
)
self.monitor.metrics_store.append(self.metrics)
常见问题和解决方案
1. 工具调用失败处理
classRobustToolManager:
def__init__(self, tools, retry_config=None):
self.tools = tools
self.retry_config = retry_config or {
'max_retries': 3,
'backoff_factor': 2,
'timeout': 30
}
defexecute_tool(self, tool_name, **kwargs):
tool = self.tools.get(tool_name)
ifnot tool:
return {"error": f"Tool '{tool_name}' not found"}
for attempt inrange(self.retry_config['max_retries']):
try:
result = self._execute_with_timeout(tool, kwargs)
return {"success": True, "result": result}
except TimeoutError:
if attempt == self.retry_config['max_retries'] - 1:
return {"error": "Tool execution timeout"}
time.sleep(self.retry_config['backoff_factor'] ** attempt)
except Exception as e:
if attempt == self.retry_config['max_retries'] - 1:
return {"error": f"Tool execution failed: {str(e)}"}
time.sleep(self.retry_config['backoff_factor'] ** attempt)
return {"error": "Max retries exceeded"}
def_execute_with_timeout(self, tool, kwargs):
import signal
deftimeout_handler(signum, frame):
raise TimeoutError("Tool execution timeout")
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(self.retry_config['timeout'])
try:
result = tool.execute(**kwargs)
signal.alarm(0) # 取消超时
return result
except Exception as e:
signal.alarm(0)
raise e
2. Token使用量控制
实际生产中，Token消耗是主要成本。需要精确控制：
import tiktoken
classTokenManager:
def__init__(self, model_name="gpt-4"):
self.encoding = tiktoken.encoding_for_model(model_name)
self.token_limits = {
'input_limit': 6000, # 输入token限制
'output_limit': 2000, # 输出token限制
'context_limit': 8000# 总上下文限制
}
defcount_tokens(self, text):
returnlen(self.encoding.encode(text))
deftruncate_context(self, context, max_tokens):
"""智能截断上下文，保留重要信息"""
current_tokens = self.count_tokens(context)
if current_tokens <= max_tokens:
return context
# 分离不同部分
parts = context.split('\n\n')
# 按重要性排序（系统提示 > 最近对话 > 历史对话）
system_parts = [p for p in parts if'System:'in p or'Task:'in p]
recent_parts = parts[-3:] # 最近3轮对话
other_parts = [p for p in parts if p notin system_parts and p notin recent_parts]
# 重新组合
result = '\n\n'.join(system_parts)
result += '\n\n' + '\n\n'.join(recent_parts)
# 如果还是超限，继续截断
whileself.count_tokens(result) > max_tokens and other_parts:
iflen(other_parts) > 0:
other_parts.pop(0)
if other_parts:
result = '\n\n'.join(system_parts) + '\n\n' + '\n\n'.join(other_parts[-2:]) + '\n\n' + '\n\n'.join(recent_parts)
return result
defoptimize_prompt(self, prompt, target_tokens):
"""优化提示词，减少token使用"""
# 移除多余空格和换行
optimized = ' '.join(prompt.split())
# 简化常见短语
replacements = {
'Please help me': 'Help',
'I would like to': 'I want to',
'Could you please': 'Please',
'Thank you very much': 'Thanks'
}
for old, new in replacements.items():
optimized = optimized.replace(old, new)
# 如果仍然超限，使用更激进的策略
ifself.count_tokens(optimized) > target_tokens:
sentences = optimized.split('. ')
# 保留前几句和后几句
iflen(sentences) > 4:
keep_sentences = sentences[:2] + sentences[-2:]
optimized = '. '.join(keep_sentences)
return optimized
3. 并发处理和队列管理
生产环境需要处理并发请求：
import asyncio
import aioredis
from typing importOptional
classAgentRequestQueue:
def__init__(self, redis_url: str, max_concurrent: int = 5):
self.redis_url = redis_url
self.max_concurrent = max_concurrent
self.semaphore = asyncio.Semaphore(max_concurrent)
self.redis_pool = None
asyncdefinit_redis(self):
self.redis_pool = await aioredis.create_redis_pool(self.redis_url)
asyncdefprocess_request(self, user_id: str, task: str, priority: int = 0):
"""处理用户请求，支持优先级"""
request_id = f"{user_id}_{int(time.time())}"
# 添加到队列
awaitself._enqueue_request(request_id, {
'user_id': user_id,
'task': task,
'priority': priority,
'timestamp': time.time()
})
# 等待处理
returnawaitself._wait_for_result(request_id)
asyncdef_enqueue_request(self, request_id: str, request_data: dict):
# 使用Redis有序集合实现优先级队列
score = -request_data['priority'] # 负数实现高优先级在前
awaitself.redis_pool.zadd('agent_queue', score, request_id)
awaitself.redis_pool.hset(f'request_{request_id}', mapping=request_data)
asyncdef_wait_for_result(self, request_id: str, timeout: int = 300):
"""等待处理结果"""
for _ inrange(timeout):
result = awaitself.redis_pool.get(f'result_{request_id}')
if result:
awaitself._cleanup_request(request_id)
return json.loads(result)
await asyncio.sleep(1)
raise TimeoutError(f"Request {request_id} timeout")
asyncdef_cleanup_request(self, request_id: str):
"""清理请求相关数据"""
awaitself.redis_pool.delete(f'request_{request_id}')
awaitself.redis_pool.delete(f'result_{request_id}')
awaitself.redis_pool.zrem('agent_queue', request_id)
classAgentWorker:
def__init__(self, agent, queue_manager):
self.agent = agent
self.queue_manager = queue_manager
self.running = False
asyncdefstart(self):
"""启动工作进程"""
self.running = True
awaitself.queue_manager.init_redis()
whileself.running:
try:
# 获取下一个请求
request_id = awaitself._get_next_request()
if request_id:
asyncwithself.queue_manager.semaphore:
awaitself._process_request(request_id)
else:
await asyncio.sleep(1) # 没有请求时短暂等待
except Exception as e:
logging.error(f"Worker error: {e}")
await asyncio.sleep(5)
asyncdef_get_next_request(self) -> Optional[str]:
"""从队列获取下一个请求"""
result = awaitself.queue_manager.redis_pool.zpopmin('agent_queue')
return result[0][0].decode() if result elseNone
asyncdef_process_request(self, request_id: str):
"""处理单个请求"""
# 获取请求数据
request_data = awaitself.queue_manager.redis_pool.hgetall(f'request_{request_id}')
ifnot request_data:
return
user_id = request_data[b'user_id'].decode()
task = request_data[b'task'].decode()
try:
# 执行Agent任务
result = awaitself.agent.solve_task(task)
# 保存结果
awaitself.queue_manager.redis_pool.set(
f'result_{request_id}',
json.dumps({
'success': True,
'result': result,
'processed_at': time.time()
}),
expire=3600# 1小时过期
)
except Exception as e:
# 保存错误结果
awaitself.queue_manager.redis_pool.set(
f'result_{request_id}',
json.dumps({
'success': False,
'error': str(e),
'processed_at': time.time()
}),
expire=3600
)
实际应用场景案例
1. 客服Agent实现
classCustomerServiceAgent:
def__init__(self):
self.knowledge_base = ProductionRAGSystem()
self.llm = ChatOpenAI(model="gpt-4")
self.conversation_memory = {}
# 预定义的工作流程
self.workflows = {
'order_inquiry': self._handle_order_inquiry,
'product_question': self._handle_product_question,
'complaint': self._handle_complaint,
'general': self._handle_general_query
}
defclassify_intent(self, user_input):
"""意图识别"""
classification_prompt = f"""
分析用户输入，判断属于以下哪种类型：
1. order_inquiry - 订单查询相关
2. product_question - 产品咨询
3. complaint - 投诉建议
4. general - 一般咨询
用户输入：{user_input}
返回分类结果（只返回类型名称）：
"""
intent = self.llm.predict(classification_prompt).strip()
return intent if intent inself.workflows else'general'
def_handle_order_inquiry(self, user_input, context):
"""处理订单查询"""
# 提取订单号
order_extraction_prompt = f"""
从用户输入中提取订单号：{user_input}
如果没有订单号，返回"NEED_ORDER_NUMBER"
如果有订单号，返回订单号
"""
order_number = self.llm.predict(order_extraction_prompt).strip()
if order_number == "NEED_ORDER_NUMBER":
return"请提供您的订单号，我来帮您查询订单状态。"
# 调用订单查询API
order_info = self._query_order_api(order_number)
if order_info:
returnf"""
您的订单信息如下：
订单号：{order_info['order_id']}
状态：{order_info['status']}
预计送达：{order_info.get('estimated_delivery', '待确定')}
还有其他需要帮助的吗？
"""
else:
return"抱歉，没有找到对应的订单信息。请确认订单号是否正确。"
def_query_order_api(self, order_number):
"""模拟订单API调用"""
# 实际实现中这里会调用真实的订单系统API
mock_orders = {
"12345": {
"order_id": "12345",
"status": "已发货",
"estimated_delivery": "2024-01-15"
}
}
return mock_orders.get(order_number)
defprocess_customer_request(self, user_id, user_input):
"""处理客服请求"""
# 获取会话历史
context = self.conversation_memory.get(user_id, [])
# 意图识别
intent = self.classify_intent(user_input)
# 根据意图调用对应处理流程
handler = self.workflows[intent]
response = handler(user_input, context)
# 更新会话历史
context.append({'user': user_input, 'assistant': response})
self.conversation_memory[user_id] = context[-10:] # 保留最近10轮对话
return response
2. 代码助手Agent
import subprocess
import tempfile
import os
classCodeAssistantAgent:
def__init__(self):
self.llm = ChatOpenAI(model="gpt-4")
self.supported_languages = ['python', 'javascript', 'bash', 'sql']
self.security_checker = CodeSecurityChecker()
defanalyze_code_request(self, user_input):
"""分析代码请求类型"""
analysis_prompt = f"""
分析用户的代码请求，返回JSON格式：
{{
"task_type": "write|debug|explain|optimize|review",
"language": "python|javascript|bash|sql|other",
"complexity": "simple|medium|complex",
"requires_execution": true|false
}}
用户请求：{user_input}
"""
try:
response = self.llm.predict(analysis_prompt)
return json.loads(response)
except:
return {
"task_type": "write",
"language": "python",
"complexity": "medium",
"requires_execution": False
}
defgenerate_code(self, requirements, language="python"):
"""生成代码"""
code_prompt = f"""
基于以下需求，生成{language}代码：
需求：{requirements}
要求：
1. 代码要完整可运行
2. 添加适当的注释
3. 包含错误处理
4. 遵循最佳实践
请只返回代码，不要额外解释：
"""
code = self.llm.predict(code_prompt)
# 清理代码格式
code = self._clean_code_response(code)
# 安全检查
ifnotself.security_checker.is_safe(code, language):
return"代码包含潜在安全风险，请修改需求后重试。"
return code
defexecute_code(self, code, language="python"):
"""安全执行代码"""
if language notinself.supported_languages:
return"不支持的编程语言"
# 安全检查
ifnotself.security_checker.is_safe(code, language):
return"代码包含不安全操作，无法执行"
try:
if language == "python":
returnself._execute_python(code)
elif language == "bash":
returnself._execute_bash(code)
# ... 其他语言
except Exception as e:
returnf"执行错误：{str(e)}"
def_execute_python(self, code):
"""在沙箱环境中执行Python代码"""
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
f.write(code)
temp_file = f.name
try:
# 使用subprocess执行，限制权限
result = subprocess.run(
['python', temp_file],
capture_output=True,
text=True,
timeout=30, # 30秒超时
cwd=tempfile.gettempdir() # 限制执行目录
)
if result.returncode == 0:
returnf"执行成功\n输出：\n{result.stdout}"
else:
returnf"执行失败\n错误：\n{result.stderr}"
finally:
os.unlink(temp_file)
def_clean_code_response(self, response):
"""清理LLM返回的代码响应"""
# 移除代码块标记
if'```'in response:
parts = response.split('```')
iflen(parts) >= 2:
code = parts[1]
# 移除语言标记
lines = code.split('\n')
if lines[0].strip() inself.supported_languages:
lines = lines[1:]
return'\n'.join(lines).strip()
return response.strip()
classCodeSecurityChecker:
def__init__(self):
self.dangerous_patterns = {
'python': [
'import os', 'import sys', 'import subprocess',
'exec(', 'eval(', 'open(', 'file(',
'__import__', 'input(', 'raw_input('
],
'bash': [
'rm -rf', 'sudo', 'chmod', 'chown',
'>', '>>', 'curl', 'wget', 'nc '
]
}
defis_safe(self, code, language):
"""检查代码是否安全"""
if language notinself.dangerous_patterns:
returnTrue
dangerous = self.dangerous_patterns[language]
code_lower = code.lower()
for pattern in dangerous:
if pattern.lower() in code_lower:
returnFalse
returnTrue
3. 数据分析Agent
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import io
import base64
classDataAnalysisAgent:
def__init__(self):
self.llm = ChatOpenAI(model="gpt-4")
self.current_dataframes = {} # 存储当前会话的数据框
defanalyze_data(self, user_query, data_source=None):
"""分析数据并生成报告"""
# 加载数据
if data_source:
df = self._load_data(data_source)
df_name = f"df_{len(self.current_dataframes)}"
self.current_dataframes[df_name] = df
# 分析查询意图
analysis_plan = self._generate_analysis_plan(user_query)
# 执行分析
results = []
for step in analysis_plan['steps']:
try:
result = self._execute_analysis_step(step)
results.append(result)
except Exception as e:
results.append(f"执行步骤 '{step}' 时出错：{str(e)}")
# 生成最终报告
report = self._generate_report(user_query, results)
return report
def_generate_analysis_plan(self, user_query):
"""生成数据分析计划"""
available_data = list(self.current_dataframes.keys())
plan_prompt = f"""
基于用户查询生成数据分析计划：
用户查询：{user_query}
可用数据：{available_data}
返回JSON格式的分析计划：
{{
"steps": [
"描述数据基本信息",
"执行具体分析",
"生成可视化图表",
"总结分析结果"
],
"required_libraries": ["pandas", "matplotlib"],
"analysis_type": "descriptive|exploratory|predictive"
}}
"""
try:
plan_response = self.llm.predict(plan_prompt)
return json.loads(plan_response)
except:
return {
"steps": ["基础数据分析", "生成统计摘要"],
"analysis_type": "descriptive"
}
def_execute_analysis_step(self, step):
"""执行分析步骤"""
ifnotself.current_dataframes:
return"没有可用的数据进行分析"
# 获取主要数据框
main_df_name = list(self.current_dataframes.keys())[0]
df = self.current_dataframes[main_df_name]
if"基本信息"in step:
returnself._get_basic_info(df)
elif"统计摘要"in step:
returnself._get_statistical_summary(df)
elif"可视化"in step:
returnself._generate_visualizations(df)
elif"相关性分析"in step:
returnself._analyze_correlations(df)
else:
returnf"未知分析步骤：{step}"
def_get_basic_info(self, df):
"""获取数据基本信息"""
info = {
"行数": len(df),
"列数": len(df.columns),
"列名": list(df.columns),
"数据类型": df.dtypes.to_dict(),
"缺失值": df.isnull().sum().to_dict()
}
returnf"""
数据基本信息：
- 数据形状：{info['行数']} 行 × {info['列数']} 列
- 列名：{', '.join(info['列名'])}
- 缺失值统计：{dict(filter(lambda x: x[1] > 0, info['缺失值'].items()))}
"""
def_get_statistical_summary(self, df):
"""生成统计摘要"""
numeric_columns = df.select_dtypes(include=['number']).columns
iflen(numeric_columns) == 0:
return"数据中没有数值型列可以进行统计分析"
summary = df[numeric_columns].describe()
# 格式化输出
summary_text = "数值列统计摘要：\n"
for col in summary.columns:
summary_text += f"\n{col}:\n"
summary_text += f" 均值: {summary.loc['mean', col]:.2f}\n"
summary_text += f" 中位数: {summary.loc['50%', col]:.2f}\n"
summary_text += f" 标准差: {summary.loc['std', col]:.2f}\n"
return summary_text
def_generate_visualizations(self, df):
"""生成可视化图表"""
numeric_columns = df.select_dtypes(include=['number']).columns
iflen(numeric_columns) == 0:
return"没有数值数据可供可视化"
# 生成分布图
fig, axes = plt.subplots(2, 2, figsize=(12, 10))
fig.suptitle('数据分布分析')
for i, col inenumerate(numeric_columns[:4]): # 最多显示4个列
row, col_idx = divmod(i, 2)
df[col].hist(ax=axes[row, col_idx], bins=20)
axes[row, col_idx].set_title(f'{col} 分布')
axes[row, col_idx].set_xlabel(col)
axes[row, col_idx].set_ylabel('频次')
# 隐藏空的子图
for i inrange(len(numeric_columns), 4):
row, col_idx = divmod(i, 2)
axes[row, col_idx].set_visible(False)
plt.tight_layout()
# 将图表转换为base64字符串
buffer = io.BytesIO()
plt.savefig(buffer, format='png')
buffer.seek(0)
image_base64 = base64.b64encode(buffer.getvalue()).decode()
plt.close()
returnf"已生成数据分布图表（base64编码）：\n[图表数据: {image_base64[:50]}...]"
def_generate_report(self, user_query, analysis_results):
"""生成最终分析报告"""
report_prompt = f"""
基于以下分析结果，生成一份专业的数据分析报告：
用户查询：{user_query}
分析结果：
{chr(10).join(analysis_results)}
请生成一份结构清晰的分析报告，包含：
1. 数据概况
2. 主要发现
3. 业务建议
4. 局限性说明
"""
report = self.llm.predict(report_prompt)
return report
调试和故障排除
常见问题诊断工具
classAgentDebugger:
def__init__(self, agent):
self.agent = agent
self.debug_logs = []
self.performance_metrics = {}
defdebug_execution(self, task, verbose=True):
"""调试Agent执行过程"""
self.debug_logs.clear()
try:
# 记录开始时间
start_time = time.time()
# 执行任务并记录每个步骤
result = self._execute_with_logging(task)
# 记录性能指标
execution_time = time.time() - start_time
self.performance_metrics = {
'execution_time': execution_time,
'total_tokens': self._count_total_tokens(),
'tool_calls': self._count_tool_calls(),
'error_count': self._count_errors()
}
if verbose:
self._print_debug_report()
return result
except Exception as e:
self.debug_logs.append({
'type': 'ERROR',
'message': str(e),
'timestamp': time.time()
})
if verbose:
self._print_error_analysis()
raise e
def_execute_with_logging(self, task):
"""执行任务并记录日志"""
self._log('TASK_START', f"开始执行任务: {task}")
# 这里需要根据实际Agent实现来记录执行步骤
# 示例：
for step inself.agent.solve_task_steps(task):
self._log('STEP', f"执行步骤: {step}")
result = self.agent.solve_task(task)
self._log('TASK_END', f"任务完成: {result}")
return result
def_log(self, log_type, message):
"""记录调试日志"""
self.debug_logs.append({
'type': log_type,
'message': message,
'timestamp': time.time()
})
def_print_debug_report(self):
"""打印调试报告"""
print("=== Agent执行调试报告 ===")
print(f"执行时间: {self.performance_metrics['execution_time']:.2f}秒")
print(f"Token使用: {self.performance_metrics['total_tokens']}")
print(f"工具调用次数: {self.performance_metrics['tool_calls']}")
print(f"错误次数: {self.performance_metrics['error_count']}")
print("\n=== 执行日志 ===")
for i, log inenumerate(self.debug_logs):
timestamp = time.strftime('%H:%M:%S', time.localtime(log['timestamp']))
print(f"[{timestamp}] {log['type']}: {log['message']}")
defanalyze_performance_bottlenecks(self):
"""分析性能瓶颈"""
bottlenecks = []
ifself.performance_metrics['execution_time'] > 30:
bottlenecks.append("执行时间过长，考虑优化推理链或并行执行")
ifself.performance_metrics['total_tokens'] > 8000:
bottlenecks.append("Token使用量过高，考虑优化提示词或截断上下文")
ifself.performance_metrics['tool_calls'] > 10:
bottlenecks.append("工具调用次数过多，检查是否存在循环调用")
return bottlenecks
结论
AI Agent开发是一个复杂的系统工程，需要在架构设计、工具集成、性能优化等多个维度进行权衡。基于Google白皮书的技术框架，我们总结几个关键点：
技术选型建议
-
- 起步阶段：使用LangChain等成熟框架快速验证概念
-
- 生产阶段：根据业务需求定制化开发，注重监控和运维
-
- 规模化阶段：考虑分布式部署、成本优化和质量保证
避免的常见陷阱
-
- 过度复杂化：不是所有任务都需要Agent，简单问题用传统LLM应用即可
-
- 忽视成本控制：Token消耗和API调用成本需要从设计阶段就考虑
-
- 缺乏监控：生产环境必须有完善的监控和日志
-
- 安全与伦理：在生产环境中，安全和伦理问题不容忽视。需要设计机制来防止Agent产生有害、不准确或不公平的输出，并确保数据隐私和安全。
大模型未来如何发展？普通人能从中受益吗？
在科技日新月异的今天，大模型已经展现出了令人瞩目的能力，从编写代码到医疗诊断，再到自动驾驶，它们的应用领域日益广泛。那么，未来大模型将如何发展？普通人又能从中获得哪些益处呢？
通用人工智能（AGI）的曙光：未来，我们可能会见证通用人工智能（AGI）的出现，这是一种能够像人类一样思考的超级模型。它们有可能帮助人类解决气候变化、癌症等全球性难题。这样的发展将极大地推动科技进步，改善人类生活。
个人专属大模型的崛起：想象一下，未来的某一天，每个人的手机里都可能拥有一个私人AI助手。这个助手了解你的喜好，记得你的日程，甚至能模仿你的语气写邮件、回微信。这样的个性化服务将使我们的生活变得更加便捷。
脑机接口与大模型的融合：脑机接口技术的发展，使得大模型与人类的思维直接连接成为可能。未来，你可能只需戴上头盔，心中想到写一篇工作总结”，大模型就能将文字直接投影到屏幕上，实现真正的心想事成。
大模型的多领域应用：大模型就像一个超级智能的多面手，在各个领域都展现出了巨大的潜力和价值。随着技术的不断发展，相信未来大模型还会给我们带来更多的惊喜。赶紧把这篇文章分享给身边的朋友，一起感受大模型的魅力吧！
那么，如何学习AI大模型？
在一线互联网企业工作十余年里，我指导过不少同行后辈，帮助他们得到了学习和成长。我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑。因此，我坚持整理和分享各种AI大模型资料，包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频。
学习阶段包括：
1.大模型系统设计
从大模型系统设计入手，讲解大模型的主要方法。包括模型架构、训练过程、优化策略等，让读者对大模型有一个全面的认识。
2.大模型提示词工程
通过大模型提示词工程，从Prompts角度入手，更好发挥模型的作用。包括提示词的构造、优化、应用等，让读者学会如何更好地利用大模型。
3.大模型平台应用开发
借助阿里云PAI平台，构建电商领域虚拟试衣系统。从需求分析、方案设计、到具体实现，详细讲解如何利用大模型构建实际应用。
4.大模型知识库应用开发
以LangChain框架为例，构建物流行业咨询智能问答系统。包括知识库的构建、问答系统的设计、到实际应用，让读者了解如何利用大模型构建智能问答系统。
5.大模型微调开发
借助以大健康、新零售、新媒体领域，构建适合当前领域的大模型。包括微调的方法、技巧、到实际应用，让读者学会如何针对特定领域进行大模型的微调。
6.SD多模态大模型
以SD多模态大模型为主，搭建文生图小程序案例。从模型选择、到小程序的设计、到实际应用，让读者了解如何利用大模型构建多模态应用。
7.大模型平台应用与开发
通过星火大模型、文心大模型等成熟大模型，构建大模型行业应用。包括行业需求分析、方案设计、到实际应用，让读者了解如何利用大模型构建行业应用。
学成之后的收获👈
• 全栈工程实现能力：通过学习，你将掌握从前端到后端，从产品经理到设计，再到数据分析等一系列技能，实现全方位的技术提升。
• 解决实际项目需求：在大数据时代，企业和机构面临海量数据处理的需求。掌握大模型应用开发技能，将使你能够更准确地分析数据，更有效地做出决策，更好地应对各种实际项目挑战。
• AI应用开发实战技能：你将学习如何基于大模型和企业数据开发AI应用，包括理论掌握、GPU算力运用、硬件知识、LangChain开发框架应用，以及项目实战经验。此外，你还将学会如何进行Fine-tuning垂直训练大模型，包括数据准备、数据蒸馏和大模型部署等一站式技能。
• 提升编码能力：大模型应用开发需要掌握机器学习算法、深度学习框架等技术，这些技术的掌握将提升你的编码能力和分析能力，使你能够编写更高质量的代码。
学习资源📚
- AI大模型学习路线图：为你提供清晰的学习路径，助你系统地掌握AI大模型知识。
- 100套AI大模型商业化落地方案：学习如何将AI大模型技术应用于实际商业场景，实现技术的商业化价值。
- 100集大模型视频教程：通过视频教程，你将更直观地学习大模型的技术细节和应用方法。
- 200本大模型PDF书籍：丰富的书籍资源，供你深入阅读和研究，拓宽你的知识视野。
- LLM面试题合集：准备面试，了解大模型领域的常见问题，提升你的面试通过率。
- AI产品经理资源合集：为你提供AI产品经理的实用资源，帮助你更好地管理和推广AI产品。
👉获取方式： 😝有需要的小伙伴，可以保存图片到wx扫描二v码免费领取【保证100%免费】
更多推荐
28
0- 0
已为社区贡献9条内容
所有评论(0)