AgentlyFormat
Advanced tools
开发说明：本项目完全使用vibe coding开发，0人工编程，从0到2.0版本（包含debug、测试集的构建以及测试）仅花费了约12小时，项目总计约15000行代码。
AgentlyFormat 让大模型 JSON 输出更稳定、更可靠，助你专注业务逻辑而非格式问题。
🚀 性能飞跃: 适配器创建速度提升81% (19.5s → 3.75s)
✅ 稳定性: 156个测试用例100%通过
🔧 架构重构: 全新的流式解析和补全引擎
🌐 API增强: 完整的REST API和WebSocket支持
📊 实时监控: 新增统计接口和健康检查
大模型在生成JSON数据时经常遇到以下问题：
AgentlyFormat 专门解决这些问题，提供稳定可靠的JSON处理能力。
pip install AgentlyFormat
from agently_format import JSONCompleter
# 创建补全器
completer = JSONCompleter()
# 不完整的JSON
incomplete_json = '{"name": "Alice", "age": 25, "skills": ["Python"'
# 智能补全
result = completer.complete(incomplete_json)
print(result.completed_json)
# 输出: {"name": "Alice", "age": 25, "skills": ["Python"]}
import asyncio
from agently_format import StreamingParser
async def parse_stream():
parser = StreamingParser()
session_id = parser.create_session()
# 模拟分块数据
chunks = [
'{"users": [',
'{"id": 1, "name": "Alice"},',
'{"id": 2, "name": "Bob"}',
'], "total": 2}'
]
for chunk in chunks:
result = await parser.parse_chunk(chunk, session_id)
print(f"进度: {result.progress:.1%}")
# 获取完整数据
final_data = parser.get_current_data(session_id)
print(final_data)
asyncio.run(parse_stream())
from agently_format import PathBuilder
builder = PathBuilder()
data = {
"api": {
"users": [
{"id": 1, "profile": {"name": "Alice"}},
{"id": 2, "profile": {"name": "Bob"}}
]
}
}
# 提取所有路径
paths = builder.build_paths(data)
print(paths)
# ['api.users.0.id', 'api.users.0.profile.name', 'api.users.1.id', 'api.users.1.profile.name']
# 通过路径获取值
value = builder.get_value_by_path(data, "api.users.0.profile.name")
print(value) # "Alice"
支持多种主流AI模型，统一的接口设计：
from agently_format.adapters import (
OpenAIAdapter, DoubaoAdapter, WenxinAdapter,
QianwenAdapter, DeepSeekAdapter, KimiAdapter
)
from agently_format.types import ModelConfig
# OpenAI适配器
openai_config = ModelConfig(
model_type="openai",
model_name="gpt-3.5-turbo",
api_key="your-api-key"
)
adapter = OpenAIAdapter(openai_config)
# 文心大模型适配器
wenxin_config = ModelConfig(
model_type="baidu",
model_name="ernie-4.0-8k",
api_key="your-api-key",
api_secret="your-api-secret"
)
wenxin_adapter = WenxinAdapter(wenxin_config)
# 千问适配器
qianwen_config = ModelConfig(
model_type="qwen",
model_name="qwen-turbo",
api_key="your-api-key"
)
qianwen_adapter = QianwenAdapter(qianwen_config)
# DeepSeek适配器
deepseek_config = ModelConfig(
model_type="deepseek",
model_name="deepseek-chat",
api_key="your-api-key"
)
deepseek_adapter = DeepSeekAdapter(deepseek_config)
# Kimi适配器
kimi_config = ModelConfig(
model_type="kimi",
model_name="moonshot-v1-8k",
api_key="your-api-key"
)
kimi_adapter = KimiAdapter(kimi_config)
# 统一的聊天补全接口
response = await adapter.chat_completion([
{"role": "user", "content": "生成一个用户信息的JSON"}
])
print(response.content)
# 启动API服务
cd AgentlyFormat
python -m agently_format.api.app
# JSON补全API
curl -X POST "http://localhost:8000/api/v1/json/complete" \
-H "Content-Type: application/json" \
-d '{"content": "{\"name\": \"Alice\", \"age\": 25", "strategy": "smart"}'
# 路径构建API
curl -X POST "http://localhost:8000/api/v1/path/build" \
-H "Content-Type: application/json" \
-d '{"data": {"user": {"name": "Alice"}}, "style": "dot"}'
class JSONCompleter:
def complete(self, json_str: str, strategy: str = "smart") -> CompletionResult:
"""补全不完整的JSON字符串"""
def validate(self, json_str: str) -> bool:
"""验证JSON格式是否正确"""
class StreamingParser:
def create_session(self, session_id: str = None) -> str:
"""创建解析会话"""
async def parse_chunk(self, chunk: str, session_id: str, is_final: bool = False) -> ParseResult:
"""解析JSON数据块"""
def get_current_data(self, session_id: str) -> dict:
"""获取当前解析的数据"""
class PathBuilder:
def build_paths(self, data: dict, style: str = "dot") -> List[str]:
"""构建数据路径列表"""
def get_value_by_path(self, data: dict, path: str) -> Any:
"""通过路径获取值"""
def convert_path(self, path: str, target_style: str) -> str:
"""转换路径格式"""
# API服务配置
AGENTLY_FORMAT_HOST=0.0.0.0
AGENTLY_FORMAT_PORT=8000
AGENTLY_FORMAT_DEBUG=false
# 模型API密钥
OPENAI_API_KEY=your-openai-key
DOUBAO_API_KEY=your-doubao-key
WENXIN_API_KEY=your-wenxin-key
WENXIN_SECRET_KEY=your-wenxin-secret
QIANWEN_API_KEY=your-qianwen-key
DEEPSEEK_API_KEY=your-deepseek-key
KIMI_API_KEY=your-kimi-key
# config.yaml
server:
host: "0.0.0.0"
port: 8000
debug: false
processing:
max_chunk_size: 1048576 # 1MB
session_ttl: 3600 # 1小时
max_sessions: 1000
models:
openai:
api_key: "${OPENAI_API_KEY}"
timeout: 30
doubao:
api_key: "${DOUBAO_API_KEY}"
timeout: 30
wenxin:
api_key: "${WENXIN_API_KEY}"
api_secret: "${WENXIN_SECRET_KEY}"
timeout: 30
qianwen:
api_key: "${QIANWEN_API_KEY}"
timeout: 30
deepseek:
api_key: "${DEEPSEEK_API_KEY}"
timeout: 30
kimi:
api_key: "${KIMI_API_KEY}"
timeout: 30
# 安装开发依赖
pip install -e ".[dev]"
# 运行测试
pytest
# 运行特定测试
pytest tests/test_core.py
# 生成覆盖率报告
pytest --cov=agently_format --cov-report=html
查看 examples/ 目录获取更多示例：
basic_usage.py - 基础功能演示streaming_example.py - 流式处理示例 (v2.0新增跨块缓冲)api_client_example.py - API客户端使用 (v2.0新增WebSocket)model_adapter_example.py - 模型适配器示例 (v2.0性能优化)advanced_usage.py - 高级功能演示 (v2.0新增差分引擎)schema_validation.py - Schema验证示例 (v2.0新增)real_time_monitoring.py - 实时监控示例 (v2.0新增)batch_processing.py - 批量处理示例 (v2.0新增)import requests
# 获取实时统计数据
response = requests.get("http://localhost:8000/api/v1/stats")
stats = response.json()
print(f"总事件数: {stats['total_events']}")
print(f"活跃会话: {stats['active_sessions']}")
import asyncio
import websockets
async def websocket_client():
uri = "ws://localhost:8000/ws"
async with websockets.connect(uri) as websocket:
await websocket.send('{"type": "parse", "data": "partial json..."}')
response = await websocket.recv()
print(f"实时响应: {response}")
asyncio.run(websocket_client())
| 指标 | v1.0.0 | v2.0.0 | 改进幅度 |
|---|---|---|---|
| 适配器创建 | 19.5s | 3.75s | 81% ⬇️ |
| 测试执行 | ~15s | 3.75s | 75% ⬇️ |
| 内存使用 | 基准 | -50% | 50% ⬇️ |
| API响应 | ~500ms | ~100ms | 80% ⬇️ |
| 并发处理 | 10 req/s | 100 req/s | 900% ⬆️ |
| 错误率 | ~5% | <0.1% | 98% ⬇️ |
欢迎贡献代码！请遵循以下步骤：
git checkout -b feature/amazing-feature)git commit -m 'Add amazing feature')git push origin feature/amazing-feature)本项目采用 Apache-2.0 许可证。
最后更新日期：2025年8月13日
AgentlyFormat - 让大模型JSON输出更稳定、更可靠！ 🚀
FAQs
专注于大模型格式化输出结果的Python库，提供稳定可靠的JSON格式化解析能力 - v2.0.0重大更新：性能提升81%，架构全面重构
We found that AgentlyFormat demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
Axios compromise traced to social engineering, showing how attacks on maintainers can bypass controls and expose the broader software supply chain.
Security News
Node.js has paused its bug bounty program after funding ended, removing payouts for vulnerability reports but keeping its security process unchanged.
Security News
The Axios compromise shows how time-dependent dependency resolution makes exposure harder to detect and contain.