PythonAI算法
基于 GPT-4o 打造 AI 智能体实战指南
基于 GPT-4o 构建 AI 智能体的完整方案。涵盖感知、规划、执行、学习四大核心模块，提供从环境配置到工具注册的基础框架代码。包含反思机制、多智能体协作及向量数据库长期记忆实现。展示数据分析、编程助手等应用场景，并给出提示词优化、流式响应及错误处理等性能技巧。最后探讨成本控制、安全性及未来多模态发展方向，帮助开发者掌握自主规划与工具调用的智能应用构建技术。
基于 GPT-4o 构建 AI 智能体的完整方案。涵盖感知、规划、执行、学习四大核心模块，提供从环境配置到工具注册的基础框架代码。包含反思机制、多智能体协作及向量数据库长期记忆实现。展示数据分析、编程助手等应用场景，并给出提示词优化、流式响应及错误处理等性能技巧。最后探讨成本控制、安全性及未来多模态发展方向，帮助开发者掌握自主规划与工具调用的智能应用构建技术。
传统的 AI 应用模式（输入 - 处理 - 输出）已经无法满足复杂场景需求。真正的智能，体现在系统能够：
而AI 智能体正是实现这一目标的技术框架。想象一下：你只需要说'帮我分析 Q2 的销售数据并做一份 PPT'，AI 就能自动完成数据查询、分析、可视化、生成报告的全流程——这就是智能体的魅力。
一个完整的智能体系统包含三大核心模块：
感知 → 规划 → 执行 → 学习 ↑ ↓ 反馈循环
1. 感知层：理解用户指令和上下文环境
2. 规划层：将目标拆解为可执行步骤（思维链、思维树）
3. 执行层：调用工具函数完成任务
4. 学习层：从历史交互中优化策略
# 安装必要库
pip install openai>=1.0.0
pip install python-dotenv
pip install requests # 用于工具调用
# 环境配置 (.env 文件)
OPENAI_API_KEY="your-api-key-here"
MODEL_NAME="gpt-4o" # 最新多模态模型
import os
from openai import OpenAI
from typing import List, Dict, Any, Callable
import json
class Tool:
"""工具基类"""
def __init__(self, name: str, description: str, func: Callable):
self.name = name
self.description = description
self.func = func
def execute(self, **kwargs):
return self.func(**kwargs)
class AIAgent:
"""AI 智能体核心类"""
def __init__(self, model: str = "gpt-4o"):
self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
self.model = model
self.tools: Dict[str, Tool] = {}
self.memory = [] # 对话记忆
self.max_steps = 10 # 最大规划步骤
def register_tool(self, tool: Tool):
"""注册工具"""
self.tools[tool.name] = tool
def think(self, prompt: str) -> List[Dict]:
"""规划任务步骤"""
planning_prompt = f"""
请将以下任务拆解为具体步骤，并选择需要使用的工具。
可用工具：{json.dumps([{'name': t.name, 'desc': t.description} for t in self.tools.values()])}
任务：{prompt}
返回 JSON 格式：
{{
"steps": [
{{
"step": 步骤序号,
"action": "执行动作描述",
"tool": "使用的工具名",
"parameters": {{参数键值对}}
}}
]
}}
"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": planning_prompt}],
response_format={"type": "json_object"}
)
plan = json.loads(response.choices[0].message.content)
return plan["steps"]
def execute_step(self, step: Dict) -> str:
"""执行单个步骤"""
if step["tool"] == "direct_response": # 直接响应无需工具
return self._generate_response(step["action"])
if step["tool"] not in self.tools:
return f"错误：找不到工具 {step['tool']}"
tool = self.tools[step["tool"]]
try:
result = tool.execute(**step.get("parameters", {}))
return str(result)
except Exception as e:
return f"工具执行失败：{str(e)}"
def run(self, task: str) -> str:
"""运行智能体"""
print(f"任务：{task}")
print("-" * 50)
# 步骤规划
steps = self.think(task)
print(f"规划步骤数：{len(steps)}")
# 顺序执行
results = []
for i, step in enumerate(steps, 1):
print(f"步骤{i}：{step['action']}")
result = self.execute_step(step)
results.append(f"步骤{i}结果：{result}")
print(f" 结果：{result[:100]}...")
# 记忆存储
self.memory.append({"step": step, "result": result})
# 汇总结果
summary = self._summarize_results(task, results)
return summary
def _generate_response(self, prompt: str) -> str:
"""生成自然语言响应"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
def _summarize_results(self, task: str, results: List[str]) -> str:
"""汇总所有步骤结果"""
summary_prompt = f"""
原始任务：{task}
执行过程：
{chr(10).join(results)}
请总结完整任务执行情况和最终结果：
"""
return self._generate_response(summary_prompt)
# 示例工具定义
def web_search(query: str, max_results: int = 3):
"""模拟网络搜索工具"""
# 实际应调用搜索引擎 API
return f"搜索'{query}'的结果：1. 相关结果 1 2. 相关结果 2 3. 相关结果 3"
def calculator(expression: str):
"""计算器工具"""
try:
return eval(expression)
except:
return "计算表达式无效"
def code_generator(requirement: str, language: str = "python"):
"""代码生成工具"""
prompt = f"用{language}编写代码：{requirement}"
# 这里调用 LLM 生成代码
return f"```{language}\n# 示例代码\nprint('Hello World')\n```"
# 初始化智能体
agent = AIAgent()
# 注册工具
agent.register_tool(Tool(
name="web_search",
description="搜索最新网络信息",
func=web_search
))
agent.register_tool(Tool(
name="calculator",
description="执行数学计算",
func=calculator
))
agent.register_tool(Tool(
name="code_generator",
description="生成指定语言的代码",
func=code_generator
))
# 运行示例
if __name__ == "__main__":
result = agent.run("请搜索最新的 AI 新闻，然后计算这些新闻的平均关注度评分")
print("\n最终结果：")
print(result)
def reflect_and_improve(self):
"""反思执行过程并优化策略"""
reflection_prompt = f"""
分析以下执行历史，提出改进建议：
{json.dumps(self.memory[-5:])}
问题：
1. 哪些步骤可以优化？
2. 是否需要额外的工具？
3. 规划逻辑有何不足？
"""
reflection = self._generate_response(reflection_prompt)
self.save_reflection(reflection) # 保存到知识库
return reflection
class MultiAgentSystem:
"""多智能体协作系统"""
def __init__(self):
self.agents = {
"planner": AIAgent(), # 规划专家
"researcher": AIAgent(), # 研究专家
"coder": AIAgent() # 编程专家
}
def collaborate(self, task: str):
"""协作完成任务"""
# 规划阶段
plan = self.agents["planner"].run(f"规划任务：{task}")
# 执行阶段
results = []
for agent_name, agent in self.agents.items():
if agent_name != "planner":
result = agent.run(f"基于以下计划执行你的部分：{plan}")
results.append(result)
# 整合阶段
final_result = self.agents["planner"].run(f"整合以下结果：{results}\n原始任务：{task}")
return final_result
import chromadb
from chromadb.config import Settings
class LongTermMemory:
"""基于向量数据库的长期记忆"""
def __init__(self):
self.client = chromadb.Client(Settings(
chroma_db_impl="duckdb+parquet",
persist_directory="./memory_db"
))
self.collection = self.client.create_collection("agent_memory")
def store_experience(self, task: str, result: str, embedding):
"""存储经验"""
self.collection.add(
embeddings=[embedding],
documents=[f"任务：{task}\n结果：{result}"],
ids=[f"exp_{len(self.collection.get())}"]
)
def retrieve_similar(self, query_embedding, n=3):
"""检索相似经验"""
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=n
)
return results["documents"]
# 自动完成：数据获取 → 清洗 → 分析 → 可视化 → 报告生成
task = """
分析公司最近一个季度的销售数据：
1. 从数据库获取 Q2 销售记录
2. 计算同比增长率
3. 识别 Top 10 产品
4. 生成可视化图表
5. 编写分析报告摘要
"""
task = """
开发一个用户注册系统：
1. 设计数据库表结构
2. 编写后端 API 接口
3. 创建前端注册表单
4. 添加邮箱验证功能
5. 编写单元测试
"""
task = """
为我制定 AI 学习计划：
1. 评估我的当前水平（已有 Python 基础）
2. 推荐学习路径
3. 提供每日学习任务
4. 每周进行知识检测
5. 根据进度调整计划
"""
# 使用思维链（Chain-of-Thought）提示
cot_prompt = """
让我们一步步思考：
首先，我需要理解用户的核心需求...
其次，我将拆解为子任务...
然后，为每个子任务选择合适工具...
最后，整合所有结果...
"""
# 实时显示思考过程
def stream_think(self, task: str):
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": task}],
stream=True
)
for chunk in response:
if chunk.choices[0].delta.content:
yield chunk.choices[0].delta.content
def execute_with_retry(self, step, max_retries=3):
for i in range(max_retries):
try:
return self.execute_step(step)
except Exception as e:
if i == max_retries - 1:
raise
print(f"第{i+1}次重试...")
self.adjust_plan(step) # 调整计划
AI 智能体代表着人工智能应用的下一个浪潮。通过本文的实战指南，你已经掌握了构建智能型 AI 应用的核心技术。从简单的工具调用到复杂的自主规划，智能体正在重新定义人机交互的边界。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online