pi-mono扩展开发全攻略:从场景需求到实战落地
pi-mono扩展开发是提升AI agent能力的关键技术,通过自定义工具和第三方API集成,开发者可以构建满足特定业务需求的智能化工作流。本文将系统讲解pi-mono扩展开发的核心概念、实现路径和最佳实践,帮助你从零开始打造专业级扩展工具。
1. 3大扩展场景解析:你需要什么样的自定义工具?
在开始开发之前,我们先明确pi-mono扩展的典型应用场景,这些场景覆盖了大部分实际开发需求:
自动化工作流场景
需求描述:开发一个能自动处理代码评审的工具,从代码仓库拉取最新提交,运行静态分析,生成评审报告。
痛点:手动执行代码评审步骤繁琐,容易遗漏关键检查点,不同项目的评审标准难以统一。
pi-mono解决方案:创建一个集成Git操作、代码分析和报告生成的复合工具,通过事件监听实现自动化触发。
验证方法:配置钩子函数在代码提交后自动运行工具,检查是否生成包含代码质量评分和改进建议的评审报告。
数据处理场景
需求描述:开发一个CSV数据转换器,能读取表格数据并输出格式化的Markdown表格。
痛点:手动转换大型CSV文件耗时且易出错,缺乏统一的数据验证机制。
pi-mono解决方案:开发支持数据验证和格式转换的工具,利用pi-mono的文件系统工具实现数据读写。
验证方法:使用包含1000行数据的测试CSV文件,检查转换后的Markdown表格格式是否正确,数据验证是否能捕获格式错误。
第三方服务集成场景
需求描述:开发地图服务集成工具,根据地址查询经纬度并计算两点间距离。
痛点:不同地图API接口差异大,密钥管理复杂,错误处理机制不统一。
pi-mono解决方案:封装地图服务API,利用pi-mono的密钥管理系统和错误处理框架。
验证方法:查询已知地址的经纬度,验证返回结果准确性;模拟API错误,检查错误处理机制是否正常工作。
图1:pi-mono交互式模式界面展示了扩展工具的使用环境,包括已加载的技能和扩展列表
⚠️ 避坑指南:
- 场景定义过于宽泛:先聚焦单一功能点实现,再逐步扩展功能边界
- 忽视用户交互流程:设计工具时需考虑在交互式环境中的使用体验
- 与内置工具功能重叠:开发前先通过
pi --list-tools检查现有工具
2. 核心概念解密:pi-mono扩展系统的3层架构
pi-mono扩展系统采用清晰的分层架构,理解这些核心概念是开发高质量扩展的基础:
工具定义层
核心功能:描述工具元信息、输入参数和执行逻辑
技术要点:每个工具必须定义唯一名称、详细描述和参数规范,执行函数接收上下文对象和参数对象。
💡 技术原理:工具定义采用声明式设计,使agent能够自动理解工具功能和使用方式,实现智能调用。
# Python工具定义示例
from pi_coding_agent import Tool, ToolContext, ParameterSchema
def create_geocoding_tool() -> Tool:
return Tool(
name="geocoding",
description="将地址转换为经纬度坐标",
parameters=ParameterSchema(
type="object",
properties={
"address": {
"type": "string",
"description": "需要转换的地址,格式为'街道,城市,国家'"
},
"lang": {
"type": "string",
"description": "返回结果的语言",
"enum": ["zh-CN", "en-US"],
"default": "zh-CN"
}
},
required=["address"]
),
execute=execute_geocoding
)
async def execute_geocoding(ctx: ToolContext, params):
# 工具执行逻辑
pass
上下文通信层
核心功能:提供工具与agent之间的信息交换机制
技术要点:通过上下文对象访问事件总线、UI交互和状态管理,实现工具间协作和用户交互。
🔧 实现方法:使用ctx.events进行事件发布和订阅,通过ctx.ui展示消息和获取用户输入,利用ctx.cache实现数据缓存。
# 上下文使用示例
async def execute_geocoding(ctx: ToolContext, params):
# 显示处理状态
await ctx.ui.show_message(f"正在查询地址: {params['address']}")
# 检查缓存
cache_key = f"geocode:{params['address']}:{params['lang']}"
cached_result = await ctx.cache.get(cache_key)
if cached_result:
return cached_result
# 执行API调用...
# 缓存结果,设置1小时过期
await ctx.cache.set(cache_key, result, ttl=3600)
# 发布事件
ctx.events.emit("geocoding:completed", {
"address": params["address"],
"coordinates": result["coordinates"]
})
return result
系统集成层
核心功能:处理API密钥管理、依赖解析和生命周期管理
技术要点:通过模型注册表获取API密钥,利用扩展加载机制管理工具生命周期,通过资源加载器处理外部依赖。
详细配置方法参见:扩展开发指南
⚠️ 避坑指南:
- 上下文对象使用不当:避免在工具执行函数外存储上下文引用,防止内存泄漏
- 参数验证缺失:始终定义完整的参数模式,确保输入数据符合预期格式
- 事件命名冲突:使用工具名作为事件前缀,如
geocoding:completed而非completed
3. 如何构建自定义工具?5步实现路径详解
按照以下步骤,你可以系统地完成一个pi-mono自定义工具的开发、测试和部署:
步骤1:创建工具目录结构
问题:工具文件如何组织才能被pi-mono正确识别和加载?
解决方案:遵循pi-mono的工具目录规范,将工具文件放在专用子目录中。
~/.pi/agent/tools/
geocoding/ # 工具目录,使用工具名称作为目录名
index.py # 工具入口文件,必须包含工具定义
helpers.py # 辅助函数模块
requirements.txt # 依赖声明文件
README.md # 工具说明文档
验证:运行pi --list-tools命令,检查工具是否出现在列表中。
步骤2:实现工具核心逻辑
问题:如何确保工具逻辑与pi-mono系统无缝集成?
解决方案:使用pi-mono提供的工具基类和上下文对象,实现符合规范的执行函数。
# index.py - 完整的地理编码工具实现
from pi_coding_agent import Tool, ToolContext, ParameterSchema
import requests
def create_geocoding_tool() -> Tool:
return Tool(
name="geocoding",
description="将地址转换为经纬度坐标",
parameters=ParameterSchema(
type="object",
properties={
"address": {
"type": "string",
"description": "需要转换的地址,格式为'街道,城市,国家'"
},
"lang": {
"type": "string",
"description": "返回结果的语言",
"enum": ["zh-CN", "en-US"],
"default": "zh-CN"
}
},
required=["address"]
),
execute=execute_geocoding
)
async def execute_geocoding(ctx: ToolContext, params):
"""
将地址转换为经纬度坐标
参数:
address: 需要转换的地址
lang: 返回结果的语言
返回:
包含经度和纬度的字典
"""
# 获取API密钥
api_key = await ctx.model_registry.get_api_key("map_service")
if not api_key:
raise ValueError("请配置地图服务API密钥")
# 构建API请求
base_url = "https://api.map.service/geocode"
params = {
"address": params["address"],
"language": params["lang"],
"key": api_key
}
try:
# 执行API调用
response = requests.get(base_url, params=params)
response.raise_for_status() # 抛出HTTP错误
data = response.json()
# 解析结果
if data["status"] != "OK":
raise ValueError(f"API错误: {data.get('error_message', '未知错误')}")
result = {
"latitude": data["results"][0]["geometry"]["location"]["lat"],
"longitude": data["results"][0]["geometry"]["location"]["lng"],
"formatted_address": data["results"][0]["formatted_address"]
}
return result
except requests.exceptions.RequestException as e:
# 处理网络错误
ctx.ui.show_error(f"请求失败: {str(e)}")
raise
步骤3:配置依赖和元数据
问题:如何管理工具所需的外部依赖?
解决方案:创建requirements.txt声明依赖,在工具目录中添加metadata.json提供额外信息。
# requirements.txt
requests>=2.25.1
python-dotenv>=0.19.0
// metadata.json
{
"version": "1.0.0",
"author": "Your Name",
"license": "MIT",
"compatibility": {
"pi-mono": ">=0.9.3"
}
}
步骤4:本地测试与调试
问题:如何验证工具功能正确性并排查问题?
解决方案:使用pi-mono的调试模式和测试工具进行验证。
# 安装工具到开发环境
ln -s ~/projects/geocoding-tool ~/.pi/agent/tools/geocoding
# 启动调试模式
pi --debug --tool geocoding
# 在交互式环境中测试
> geocoding {"address": "北京市海淀区中关村南大街5号"}
验证方法:检查返回结果是否符合预期,测试边界情况(如无效地址、网络错误等)。
步骤5:打包与分发
问题:如何分享开发的工具给其他用户?
解决方案:将工具打包为Python包,通过pip分发或提交到pi-mono扩展市场。
# setup.py
from setuptools import setup, find_packages
setup(
name="pi-geocoding-tool",
version="1.0.0",
packages=find_packages(),
entry_points={
"pi.tools": [
"geocoding = geocoding.index:create_geocoding_tool"
]
},
install_requires=[
"requests>=2.25.1",
"python-dotenv>=0.19.0"
]
)
⚠️ 避坑指南:
- 硬编码API密钥:始终使用
model_registry.get_api_key()获取密钥- 缺少错误处理:必须处理网络错误、API错误和无效输入等异常情况
- 忽略版本兼容性:在metadata中明确指定兼容的pi-mono版本范围
4. 进阶优化:提升工具性能和可靠性的7个技巧
开发基础工具后,通过以下优化技巧可以显著提升工具质量和用户体验:
实现智能缓存策略
问题:重复请求相同数据浪费API配额和响应时间
解决方案:使用上下文缓存API响应,设置合理的过期时间
# 智能缓存实现
async def execute_geocoding(ctx: ToolContext, params):
# 生成缓存键,包含所有参数
cache_key = f"geocode:{hash(frozenset(params.items()))}"
# 尝试获取缓存
cached_result = await ctx.cache.get(cache_key)
if cached_result:
# 添加缓存标记,帮助用户识别
cached_result["from_cache"] = True
return cached_result
# 执行API调用...
# 根据地址类型设置不同缓存时间
# 静态地址(如知名建筑)缓存时间长,动态地址(如临时活动场所)缓存时间短
if "临时" in params["address"] or "活动" in params["address"]:
ttl = 3600 # 1小时
else:
ttl = 86400 # 24小时
await ctx.cache.set(cache_key, result, ttl=ttl)
return result
实现异步处理机制
问题:长时间运行的任务会阻塞agent响应
解决方案:使用异步处理和事件通知机制
# 异步处理实现
async def execute_geocoding_batch(ctx: ToolContext, params):
"""批量地址解析工具"""
addresses = params["addresses"]
task_id = str(uuid.uuid4()) # 生成唯一任务ID
# 立即返回任务ID
ctx.ui.show_message(f"批量解析任务已启动,任务ID: {task_id}")
# 在后台处理任务
async def process_batch():
results = []
for address in addresses:
# 调用单个地址解析
result = await execute_geocoding(ctx, {"address": address})
results.append({
"address": address,
"result": result
})
# 发送进度事件
ctx.events.emit("geocoding:batch_progress", {
"task_id": task_id,
"progress": len(results)/len(addresses),
"completed": len(results),
"total": len(addresses)
})
# 发送完成事件
ctx.events.emit("geocoding:batch_complete", {
"task_id": task_id,
"results": results
})
# 启动后台任务
ctx.background_tasks.add(process_batch())
return {
"task_id": task_id,
"message": "批量解析已启动,请监听事件获取结果"
}
实现细粒度错误处理
问题:单一错误导致整个工具执行失败
解决方案:分类处理不同类型错误,提供恢复机制
# 细粒度错误处理
async def execute_geocoding(ctx: ToolContext, params):
try:
api_key = await ctx.model_registry.get_api_key("map_service")
if not api_key:
# 可恢复错误:提示用户配置密钥
raise RecoverableError(
"缺少地图服务API密钥",
"请在设置中配置MAP_SERVICE_API_KEY",
"settings://api-keys" # 可直接跳转的设置页面
)
# API调用...
except requests.exceptions.ConnectionError:
# 可重试错误
raise RetryableError(
"网络连接失败",
"无法连接到地图服务,请检查网络连接",
retry_after=10 # 建议10秒后重试
)
except requests.exceptions.Timeout:
# 可重试错误
raise RetryableError(
"请求超时",
"地图服务响应超时,请稍后重试",
retry_after=15
)
except ValueError as e:
# 不可恢复错误
raise ToolError(
"地址解析失败",
str(e)
)
except Exception as e:
# 未知错误
ctx.logger.error(f"地理编码工具未知错误: {str(e)}", exc_info=True)
raise ToolError(
"工具执行失败",
"发生未知错误,请查看日志获取详细信息"
)
图2:pi-mono会话树视图展示了工具调用历史和上下文切换,帮助追踪工具执行流程
⚠️ 避坑指南:
- 缓存策略不当:避免对频繁变化数据设置过长缓存时间
- 缺少进度反馈:长时间运行的工具必须提供进度更新
- 错误信息不明确:错误消息应包含问题原因和解决建议
- 忽略资源清理:使用
try...finally确保临时资源正确释放
5. 地图服务集成实战:从API对接到底层优化
以地图服务集成为例,我们将完整实现一个功能完善的地理编码工具,涵盖API对接、错误处理、性能优化等关键环节。
需求分析与设计
核心功能:
- 将地址转换为经纬度坐标
- 计算两个坐标点之间的距离
- 支持批量地址解析
- 缓存常用地址解析结果
API选择:采用高德地图API,提供稳定的地理编码服务
完整实现代码
# ~/.pi/agent/tools/geocoding/index.py
from pi_coding_agent import Tool, ToolContext, ParameterSchema, RecoverableError
from pi_coding_agent.utils import truncate_text
import requests
import hashlib
import uuid
from typing import Dict, Any, List
def create_geocoding_tools() -> List[Tool]:
"""创建地理编码相关工具集合"""
return [
create_geocode_tool(),
create_distance_calculator_tool(),
create_batch_geocode_tool()
]
def create_geocode_tool() -> Tool:
"""创建地址转坐标工具"""
return Tool(
name="geocode",
description="将地址转换为经纬度坐标",
parameters=ParameterSchema(
type="object",
properties={
"address": {
"type": "string",
"description": "需要转换的地址,格式为'街道,城市,国家'"
},
"city": {
"type": "string",
"description": "可选,指定城市以提高精度"
},
"lang": {
"type": "string",
"description": "返回结果的语言",
"enum": ["zh-CN", "en-US"],
"default": "zh-CN"
}
},
required=["address"]
),
execute=execute_geocode
)
def create_distance_calculator_tool() -> Tool:
"""创建距离计算工具"""
return Tool(
name="calculate_distance",
description="计算两个经纬度坐标之间的距离",
parameters=ParameterSchema(
type="object",
properties={
"origin": {
"type": "object",
"description": "起点坐标",
"properties": {
"lat": {"type": "number", "description": "纬度"},
"lng": {"type": "number", "description": "经度"}
},
"required": ["lat", "lng"]
},
"destination": {
"type": "object",
"description": "终点坐标",
"properties": {
"lat": {"type": "number", "description": "纬度"},
"lng": {"type": "number", "description": "经度"}
},
"required": ["lat", "lng"]
},
"unit": {
"type": "string",
"description": "距离单位",
"enum": ["km", "m", "mi"],
"default": "km"
}
},
required=["origin", "destination"]
),
execute=execute_calculate_distance
)
def create_batch_geocode_tool() -> Tool:
"""创建批量地址解析工具"""
return Tool(
name="batch_geocode",
description="批量将地址转换为经纬度坐标",
parameters=ParameterSchema(
type="object",
properties={
"addresses": {
"type": "array",
"description": "地址列表",
"items": {"type": "string"}
},
"lang": {
"type": "string",
"description": "返回结果的语言",
"enum": ["zh-CN", "en-US"],
"default": "zh-CN"
}
},
required=["addresses"]
),
execute=execute_batch_geocode
)
async def execute_geocode(ctx: ToolContext, params: Dict[str, Any]) -> Dict[str, Any]:
"""
将地址转换为经纬度坐标
为什么这么做:
1. 使用缓存减少API调用,降低延迟和成本
2. 详细的错误分类帮助用户快速定位问题
3. 结构化返回结果便于后续处理
"""
# 生成缓存键
cache_key = f"geocode:{hashlib.md5(str(sorted(params.items())).encode()).hexdigest()}"
# 检查缓存
cached_result = await ctx.cache.get(cache_key)
if cached_result:
cached_result["from_cache"] = True
return cached_result
# 获取API密钥
api_key = await ctx.model_registry.get_api_key("amap")
if not api_key:
raise RecoverableError(
"缺少高德地图API密钥",
"请在设置中配置AMAP_API_KEY",
"settings://api-keys"
)
# 构建请求参数
request_params = {
"key": api_key,
"address": params["address"],
"output": "json",
"language": params.get("lang", "zh-CN")
}
if "city" in params and params["city"]:
request_params["city"] = params["city"]
try:
# 执行API调用
response = requests.get(
"https://restapi.amap.com/v3/geocode/geo",
params=request_params,
timeout=10
)
response.raise_for_status()
data = response.json()
# 解析结果
if data["status"] != "1":
error_msg = data.get("info", "未知错误")
error_code = data.get("infocode", "未知错误码")
raise ValueError(f"API错误: {error_msg} (错误码: {error_code})")
if not data["geocodes"]:
raise ValueError("未找到匹配的地址")
geocode = data["geocodes"][0]
location = geocode["location"].split(",")
result = {
"formatted_address": geocode["formatted_address"],
"province": geocode.get("province", ""),
"city": geocode.get("city", ""),
"district": geocode.get("district", ""),
"longitude": float(location[0]),
"latitude": float(location[1]),
"adcode": geocode.get("adcode", ""),
"level": geocode.get("level", "")
}
# 根据地址级别设置缓存时间
# 级别越高(如门牌号)地址越稳定,缓存时间越长
level_weights = {"门牌号": 7, "街道": 5, "乡镇": 3, "区县": 2, "城市": 1, "省份": 1}
ttl = level_weights.get(result["level"], 2) * 86400 # 基础单位为天
# 缓存结果
await ctx.cache.set(cache_key, result, ttl=ttl)
return result
except requests.exceptions.ConnectionError:
raise RecoverableError(
"网络连接失败",
"无法连接到地图服务,请检查网络连接",
retry_after=10
)
except requests.exceptions.Timeout:
raise RecoverableError(
"请求超时",
"地图服务响应超时,请稍后重试",
retry_after=15
)
async def execute_calculate_distance(ctx: ToolContext, params: Dict[str, Any]) -> Dict[str, Any]:
"""
计算两个坐标点之间的距离
为什么这么做:
1. 实现本地距离计算避免API调用,降低延迟和成本
2. 支持多种单位满足不同场景需求
3. 使用Haversine公式保证计算精度
"""
from math import radians, sin, cos, sqrt, atan2
# 提取坐标
origin = params["origin"]
destination = params["destination"]
unit = params.get("unit", "km")
# 地球半径(公里)
R = 6371.0
# 转换为弧度
lat1, lon1 = radians(origin["lat"]), radians(origin["lng"])
lat2, lon2 = radians(destination["lat"]), radians(destination["lng"])
# Haversine公式
dlon = lon2 - lon1
dlat = lat2 - lat1
a = sin(dlat / 2)**2 + cos(lat1) * cos(lat2) * sin(dlon / 2)**2
c = 2 * atan2(sqrt(a), sqrt(1 - a))
distance_km = R * c
# 转换单位
if unit == "m":
distance = distance_km * 1000
elif unit == "mi":
distance = distance_km * 0.621371
else: # km
distance = distance_km
return {
"distance": round(distance, 2),
"unit": unit,
"origin": origin,
"destination": destination
}
async def execute_batch_geocode(ctx: ToolContext, params: Dict[str, Any]) -> Dict[str, Any]:
"""
批量地址解析
为什么这么做:
1. 异步处理批量请求提高效率
2. 单独处理每个地址的错误,避免一个失败导致全部失败
3. 提供进度反馈提升用户体验
"""
addresses = params["addresses"]
lang = params.get("lang", "zh-CN")
task_id = str(uuid.uuid4())
total = len(addresses)
# 立即返回任务ID
ctx.ui.show_message(f"批量地址解析已启动,共{total}个地址,任务ID: {task_id}")
# 后台处理函数
async def process_batch():
results = []
for i, address in enumerate(addresses):
try:
# 调用单个地址解析
result = await execute_geocode(ctx, {
"address": address,
"lang": lang
})
results.append({
"address": address,
"result": result,
"success": True
})
except Exception as e:
# 记录错误但继续处理其他地址
results.append({
"address": address,
"error": str(e),
"success": False
})
# 发送进度更新
progress = (i + 1) / total
ctx.events.emit("batch_geocode:progress", {
"task_id": task_id,
"progress": progress,
"completed": i + 1,
"total": total,
"address": address
})
# 发送完成事件
ctx.events.emit("batch_geocode:complete", {
"task_id": task_id,
"results": results,
"success_count": sum(1 for r in results if r["success"]),
"total": total
})
# 添加到后台任务
ctx.background_tasks.add(process_batch())
return {
"task_id": task_id,
"message": f"批量解析已启动,共{total}个地址",
"progress_url": f"events://batch_geocode:progress?task_id={task_id}"
}
工具使用示例
# 安装工具
git clone https://gitcode.com/GitHub_Trending/pi/pi-mono
cd pi-mono
ln -s $(pwd)/examples/extensions/geocoding ~/.pi/agent/tools/
# 配置API密钥
pi settings set apiKeys.amap "your_amap_api_key"
# 在pi-mono交互式模式中使用
pi
# 地址转坐标
> geocode {"address": "北京市海淀区中关村南大街5号", "city": "北京市"}
# 计算距离
> calculate_distance {
"origin": {"lat": 39.990475, "lng": 116.313596},
"destination": {"lat": 39.908823, "lng": 116.397470},
"unit": "km"
}
# 批量解析
> batch_geocode {
"addresses": [
"上海市浦东新区张江高科技园区",
"广州市天河区珠江新城",
"深圳市南山区科技园"
]
}
性能优化点
- 多级缓存策略:根据地址稳定性动态调整缓存时间
- 本地距离计算:避免调用外部API,降低延迟和成本
- 错误隔离机制:批量处理时单个地址错误不影响整体任务
- 进度反馈系统:通过事件机制提供实时进度更新
⚠️ 避坑指南:
- API密钥安全风险:避免在代码中硬编码密钥,使用pi-mono密钥管理系统
- 批量请求频率控制:添加请求间隔避免触发API速率限制
- 地址格式标准化:在工具内部对输入地址进行标准化处理
- 结果数据过大:对返回结果进行适当截断,使用
truncate_text工具
总结
pi-mono扩展开发是一项强大的技术,通过自定义工具和第三方API集成,可以显著扩展AI agent的能力边界。本文从需求场景出发,详细介绍了pi-mono扩展开发的核心概念、实现路径、进阶优化和实战案例,涵盖了工具开发的全生命周期。
关键要点回顾:
- 采用"需求场景→核心概念→实现路径→进阶优化→实战案例"的开发流程
- 遵循pi-mono的工具目录规范和API设计模式
- 充分利用上下文对象实现工具间通信和状态管理
- 采用安全的密钥管理策略保护敏感信息
- 通过缓存、异步处理和错误隔离提升工具性能和可靠性
通过本文介绍的方法和技巧,你可以开发出高质量的pi-mono扩展工具,满足特定业务需求,构建个性化的AI工作流。扩展开发是一个持续迭代的过程,建议从简单功能开始,逐步完善和优化,最终打造专业级的扩展工具。
详细开发文档参见:扩展开发指南
GLM-5智谱 AI 正式发布 GLM-5,旨在应对复杂系统工程和长时域智能体任务。Jinja00
GLM-5-w4a8GLM-5-w4a8基于混合专家架构,专为复杂系统工程与长周期智能体任务设计。支持单/多节点部署,适配Atlas 800T A3,采用w4a8量化技术,结合vLLM推理优化,高效平衡性能与精度,助力智能应用开发Jinja00
jiuwenclawJiuwenClaw 是一款基于openJiuwen开发的智能AI Agent,它能够将大语言模型的强大能力,通过你日常使用的各类通讯应用,直接延伸至你的指尖。Python0220- QQwen3.5-397B-A17BQwen3.5 实现了重大飞跃,整合了多模态学习、架构效率、强化学习规模以及全球可访问性等方面的突破性进展,旨在为开发者和企业赋予前所未有的能力与效率。Jinja00
AtomGit城市坐标计划AtomGit 城市坐标计划开启!让开源有坐标,让城市有星火。致力于与城市合伙人共同构建并长期运营一个健康、活跃的本地开发者生态。01
AntSK基于.Net9 + AntBlazor + SemanticKernel 和KernelMemory 打造的AI知识库/智能体,支持本地离线AI大模型。可以不联网离线运行。支持aspire观测应用数据CSS01

