共计 2224 个字符，预计需要花费 6 分钟才能阅读完成。
核心概念
智能代理（Agent）是一种能够感知环境、自主决策并执行动作的软件实体。它通常由三个核心组件构成：
- 感知模块：负责收集环境信息，例如通过 API 获取数据或解析用户输入
- 决策模块：基于当前状态和目标进行分析判断，常依赖预定义规则或机器学习模型
- 执行模块：将决策转化为具体操作，如调用外部服务或生成响应
典型应用场景包括：
- 自动化客服对话系统
- 智能数据分析助手
- 物联网设备控制中枢
环境准备
构建 Python Agent 需要以下工具链：
pip install openai langchain python-dotenv aiohttp关键库说明：
langchain：提供 Agent 开发框架openai：接入大语言模型 APIaiohttp：异步 HTTP 请求处理python-dotenv：管理环境变量
代码实现：天气查询 Agent
以下是一个具备完整功能的天气 Agent 示例：
import os
import aiohttp
from dotenv import load_dotenv
from typing import Dict, Any
load_dotenv()
class WeatherAgent:
def __init__(self):
self.api_key = os.getenv('WEATHER_API_KEY')
self.base_url = "https://api.weatherapi.com/v1/current.json"
async def _fetch_weather(self, location: str) -> Dict[str, Any]:
params = {'key': self.api_key, 'q': location}
async with aiohttp.ClientSession() as session:
async with session.get(self.base_url, params=params) as resp:
if resp.status != 200:
raise ValueError(f"API 请求失败: {resp.status}")
return await resp.json()
async def run(self, user_input: str) -> str:
"""
决策流程：1. 解析输入中的地理位置
2. 调用天气 API
3. 格式化返回结果
"""
try:
data = await self._fetch_weather(user_input)
current = data['current']
return (f"{user_input}当前天气:\n"
f"温度: {current['temp_c']}°C\n"
f"条件: {current['condition']['text']}"
)
except Exception as e:
return f"查询失败: {str(e)}"
# 使用示例
async def main():
agent = WeatherAgent()
print(await agent.run("北京"))
import asyncio
asyncio.run(main())代码关键点说明：
- 通过
aiohttp实现异步 API 调用 - 使用环境变量管理敏感 API 密钥
- 明确的错误处理流程
- 结构化数据解析与格式化输出
避坑指南
常见问题与解决方案
- API 限流错误：
- 现象：收到 429 状态码
- 解决：实现请求队列或添加 retry 机制
from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3))
async def _fetch_weather(self, location: str):
# 原有实现...- 异步阻塞问题：
- 现象：程序卡死无响应
-
解决：避免在异步函数中调用同步 IO 操作
-
环境变量加载失败：
- 确保.env 文件与脚本同目录
- 变量名大小写敏感
进阶建议
添加记忆功能
通过维护对话历史实现上下文感知：
class MemoryAgent(WeatherAgent):
def __init__(self):
super().__init__()
self.conversation_history = []
async def run(self, user_input: str) -> str:
self.conversation_history.append(f"用户: {user_input}")
response = await super().run(user_input)
self.conversation_history.append(f"系统: {response}")
return response性能优化技巧
- 请求批处理：对多个地点查询合并 API 请求
- 缓存机制：对重复查询使用本地缓存
- 连接池配置：优化 aiohttp 的 TCP 连接复用
from aiohttp import TCPConnector
async with aiohttp.ClientSession(connector=TCPConnector(limit=10)
) as session:
# 使用连接池...实践思考
- 如何扩展 Agent 支持多工具切换（如天气查询 + 航班搜索）？
- 当需要处理长时间运行任务时，应该如何设计状态保持机制？
- 在分布式环境中部署 Agent 时，需要考虑哪些特殊的架构问题？
通过这个基础示例，你已经掌握了 Agent 开发的核心模式。接下来可以尝试集成更复杂的决策逻辑，或将其接入实际业务系统。记住良好的错误处理和日志记录是生产级 Agent 的关键要素。
正文完
