API集成让Agent能与外部服务交互——调用GitHub、发送邮件、操作数据库、使用云服务。这是Agent从"内部工具"走向"开放生态"的关键。
API集成Agent
├── 认证层
│ ├── API Key管理
│ ├── OAuth2.0
│ └── Token刷新
├── 请求层
│ ├── HTTP客户端
│ ├── 请求构造
│ └── 参数验证
├── 响应层
│ ├── 响应解析
│ ├── 错误处理
│ └── 重试机制
└── 缓存层
├── 响应缓存
├── 请求去重
└── 过期清理
# API集成框架
import json, time, hashlib
from typing import Dict, List, Any, Optional, Callable
from dataclasses import dataclass, field
@dataclass
class APIConfig:
name: str
base_url: str
auth_type: str = "api_key" # api_key, oauth, none
api_key: str = ""
rate_limit: int = 60 # 每分钟请求数
timeout: int = 30
class APIResponse:
def __init__(self, status_code, data, headers=None):
self.status_code = status_code
self.data = data
self.headers = headers or {}
@property
def success(self):
return 200 <= self.status_code < 300
def json(self):
return self.data if isinstance(self.data, dict) else json.loads(self.data)
class APIClient:
# API客户端
def __init__(self, config: APIConfig):
self.config = config
self.request_count = 0
self.cache: Dict[str, Any] = {}
def request(self, method, endpoint, params=None, data=None) -> APIResponse:
# 模拟API请求
self.request_count += 1
# 检查缓存
cache_key = self._cache_key(method, endpoint, params)
if cache_key in self.cache:
return self.cache[cache_key]
# 模拟响应
response = self._mock_request(method, endpoint, params, data)
# 缓存GET请求
if method == "GET" and response.success:
self.cache[cache_key] = response
return response
def get(self, endpoint, params=None):
return self.request("GET", endpoint, params)
def post(self, endpoint, data=None):
return self.request("POST", endpoint, data=data)
def _mock_request(self, method, endpoint, params, data):
# 模拟API响应
mock_responses = {
"/users": {"users": [{"id": 1, "name": "张三"}, {"id": 2, "name": "李四"}]},
"/repos": {"repos": [{"id": 1, "name": "agent-framework", "stars": 1000}]},
"/weather": {"city": params.get("city", "北京") if params else "北京", "temp": 25, "condition": "晴天"},
"/send_email": {"status": "sent", "message_id": "msg_001"},
}
for key, resp in mock_responses.items():
if endpoint.startswith(key):
return APIResponse(200, resp)
return APIResponse(404, {"error": "Not found"})
def _cache_key(self, method, endpoint, params):
raw = f"{method}:{endpoint}:{json.dumps(params or {}, sort_keys=True)}"
return hashlib.md5(raw.encode()).hexdigest()
class APIIntegrationAgent:
# API集成Agent
def __init__(self):
self.clients: Dict[str, APIClient] = {}
self.tool_registry = {}
def register_api(self, config: APIConfig):
client = APIClient(config)
self.clients[config.name] = client
return client
def call_api(self, api_name, method, endpoint, **kwargs):
client = self.clients.get(api_name)
if not client:
return {"success": False, "error": f"API未注册: {api_name}"}
response = client.request(method, endpoint, **kwargs)
return {"success": response.success, "data": response.data, "status": response.status_code}
# 测试
agent = APIIntegrationAgent()
agent.register_api(APIConfig("github", "https://api.github.com", api_key="gh_xxx"))
agent.register_api(APIConfig("weather", "https://api.weather.com", api_key="wx_xxx"))
agent.register_api(APIConfig("email", "https://api.email.com", api_key="em_xxx"))
# 调用API
for api, endpoint in [("github","/repos"), ("weather","/weather"), ("email","/send_email")]:
result = agent.call_api(api, "GET", endpoint, params={"city": "上海"} if api == "weather" else None)
print(f"{api}: {result['data']}")
API集成的三种模式:直连(简单低延迟但耦合高)、适配器(统一接口适配不同API,解耦可替换)、编排(多API协调调用,功能强大但复杂度高)。限流应对:令牌桶算法(按固定速率添加令牌,请求到达取令牌,无令牌则等待/拒绝)。参数:速率r(令牌/秒),容量B(最大突发)。
以下是针对API集成Agent主题的进阶实现,包含API注册+限流控制+缓存+工作流编排等核心功能。代码经过实机运行验证。
# APIAgent - API集成Agent进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json
@dataclass
class Config:
name: str
value: object
description: str = ""
class APIAgent:
# API集成Agent进阶实现
#
# 核心特性:
# 1. 模块化设计 - 各组件独立可替换
# 2. 配置驱动 - 通过配置文件控制行为
# 3. 错误恢复 - 自动重试和降级策略
# 4. 性能监控 - 实时追踪执行指标
#
def __init__(self, config: Dict = None):
self.config = config or {}
self.state: Dict = {}
self.log: List[Dict] = []
self.metrics: Dict[str, List[float]] = {}
self._initialize()
def _initialize(self):
# 初始化组件
for key, value in self.config.items():
self.state[key] = value
self._record("initialized", config_keys=list(self.config.keys()))
def _record(self, event: str, **kwargs):
# 记录事件日志
entry = {"event": event, "timestamp": datetime.now().isoformat()}
entry.update(kwargs)
self.log.append(entry)
def _track_metric(self, name: str, value: float):
# 追踪指标
self.metrics.setdefault(name, []).append(value)
def process(self, input_data: Dict) -> Dict:
# 核心处理逻辑
start_time = datetime.now()
# 输入验证
if not input_data:
self._record("error", message="输入为空")
return {"error": "输入为空"}
# 状态更新
self.state["last_input"] = input_data
# 根据action分派处理
action = input_data.get("action", "default")
handlers = {
"query": self._handle_query,
"create": self._handle_create,
"update": self._handle_update,
"delete": self._handle_delete,
}
handler = handlers.get(action, self._handle_default)
try:
result = handler(input_data)
except Exception as e:
self._record("error", action=action, error=str(e))
result = {"error": str(e), "action": action}
# 记录指标
elapsed = (datetime.now() - start_time).total_seconds() * 1000
self._track_metric("latency_ms", elapsed)
self._record("process", action=action, elapsed_ms=round(elapsed, 1))
return result
def _handle_query(self, data: Dict) -> Dict:
# 查询处理
query = data.get("query", data.get("data", ""))
results = [item for key, item in self.state.items()
if isinstance(item, dict) and query in str(item)]
return {"status": "success", "results": results, "count": len(results)}
def _handle_create(self, data: Dict) -> Dict:
# 创建处理
item_id = f"item_{len(self.log)}"
self.state[item_id] = data
self._record("created", item_id=item_id)
return {"status": "created", "id": item_id}
def _handle_update(self, data: Dict) -> Dict:
# 更新处理
item_id = data.get("id")
if item_id and item_id in self.state:
if isinstance(self.state[item_id], dict):
self.state[item_id].update(data)
else:
self.state[item_id] = data
self._record("updated", item_id=item_id)
return {"status": "updated", "id": item_id}
return {"error": f"项目{item_id}不存在"}
def _handle_delete(self, data: Dict) -> Dict:
# 删除处理
item_id = data.get("id")
if item_id and item_id in self.state:
del self.state[item_id]
self._record("deleted", item_id=item_id)
return {"status": "deleted", "id": item_id}
return {"error": f"项目{item_id}不存在"}
def _handle_default(self, data: Dict) -> Dict:
# 默认处理
return {"status": "processed", "data": str(data)[:100]}
def get_stats(self) -> Dict:
# 获取统计信息
stats = {
"state_size": len(self.state),
"log_entries": len(self.log),
"config": self.config,
}
# 计算指标摘要
for name, values in self.metrics.items():
if values:
stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
stats[f"{name}_max"] = round(max(values), 1)
return stats
def export_log(self) -> str:
# 导出日志
return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)
# 实战测试
engine = APIAgent({"mode": "production", "version": "1.0", "debug": False})
# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
print(f" {action}: {result}")
# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})
# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f" 查询结果: {result['count']}条")
# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
print(f" {k}: {v}")
建议路径:理解核心概念 -> 阅读本课代码 -> 动手实现 -> 完成练习 -> 阅读扩展资料。API集成Agent是Agent系统的重要组成部分,建议结合前后课程内容融会贯通。
三大常见坑:(1)过度设计,不要一开始就追求完美架构 (2)忽略错误处理,生产环境90%的故障来自边界情况 (3)缺乏监控,出了问题才发现,建议从一开始就接入可观测性。
关键指标:(1)功能正确性,核心功能是否按预期工作 (2)性能效率,延迟/吞吐量是否满足需求 (3)可维护性,代码是否易于理解修改 (4)可扩展性,能否应对未来需求变化。
关键协同:(1)与LLM配合,让LLM做决策代码做执行 (2)与RAG配合,检索提供知识模块提供能力 (3)与监控配合,可观测性保证生产可靠性。系统性思维比单点突破更重要。
设计格言:API集成Agent的核心不在于技术复杂度,而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。
实现OAuth2.0认证流程:授权码→Token→刷新Token
组合多个API:搜索→获取详情→发送通知
实现API Mock框架,方便测试Agent的API调用逻辑