open-webui/Fu-Jie_openwebui-extensions
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(infographic): release v1.5.0 with smart language detection & organize debug tools

Browse Source
This commit is contained in:
fujie
2026-01-28 02:14:30 +08:00
parent e412aeb93d
commit 219ba83df3
24 changed files with 6320 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

568
plugins/debug/github-copilot-sdk/guides/COPILOT_TOOLS_QUICKSTART.md Normal file
View File

@@ -0,0 +1,568 @@
# GitHub Copilot SDK 自定义工具快速入门
## 🎯 目标
在 OpenWebUI Pipe 中直接使用 GitHub Copilot SDK 的自定义工具功能，无需集成 OpenWebUI Function 系统。
---
## 📖 基础概念
### Copilot SDK Tool 的三要素
```python
from copilot.types import Tool, ToolInvocation, ToolResult
# 1. Tool Definition（工具定义）
tool = Tool(
name="tool_name", # 工具名称
description="What it does", # 描述（给 AI 看的）
parameters={...}, # JSON Schema 参数定义
handler=handler_function # 处理函数
)
# 2. Tool Handler（处理函数）
async def handler_function(invocation: ToolInvocation) -> ToolResult:
# invocation 包含：
# - session_id: 会话 ID
# - tool_call_id: 调用 ID
# - tool_name: 工具名称
# - arguments: dict（实际参数）
result = do_something(invocation["arguments"])
return ToolResult(
textResultForLlm="结果文本",
resultType="success", # 或 "failure"
error=None,
toolTelemetry={}
)
# 3. Session Configuration（会话配置）
session_config = SessionConfig(
model="claude-sonnet-4.5",
tools=[tool1, tool2, tool3], # ✅ 传入工具列表
streaming=True
)
```
---
## 💻 完整实现示例
### 示例 1：获取当前时间
```python
from datetime import datetime
from copilot.types import Tool, ToolInvocation, ToolResult
def create_time_tool():
"""创建获取时间的工具"""
async def get_time_handler(invocation: ToolInvocation) -> ToolResult:
"""工具处理函数"""
try:
# 获取参数
timezone = invocation["arguments"].get("timezone", "UTC")
format_str = invocation["arguments"].get("format", "%Y-%m-%d %H:%M:%S")
# 执行逻辑
current_time = datetime.now().strftime(format_str)
result_text = f"Current time: {current_time}"
# 返回结果
return ToolResult(
textResultForLlm=result_text,
resultType="success",
error=None,
toolTelemetry={"execution_time": "fast"}
)
except Exception as e:
return ToolResult(
textResultForLlm=f"Error getting time: {str(e)}",
resultType="failure",
error=str(e),
toolTelemetry={}
)
# 创建工具定义
return Tool(
name="get_current_time",
description="Get the current date and time. Useful when user asks 'what time is it' or needs to know the current date.",
parameters={
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "Timezone name (e.g., 'UTC', 'Asia/Shanghai')",
"default": "UTC"
},
"format": {
"type": "string",
"description": "Time format string",
"default": "%Y-%m-%d %H:%M:%S"
}
}
},
handler=get_time_handler
)
```
### 示例 2：数学计算器
```python
def create_calculator_tool():
"""创建计算器工具"""
async def calculate_handler(invocation: ToolInvocation) -> ToolResult:
try:
expression = invocation["arguments"].get("expression", "")
# 安全检查
allowed_chars = set("0123456789+-*/()., ")
if not all(c in allowed_chars for c in expression):
raise ValueError("Expression contains invalid characters")
# 计算（安全的 eval）
result = eval(expression, {"__builtins__": {}})
return ToolResult(
textResultForLlm=f"The result of {expression} is {result}",
resultType="success",
error=None,
toolTelemetry={}
)
except Exception as e:
return ToolResult(
textResultForLlm=f"Calculation error: {str(e)}",
resultType="failure",
error=str(e),
toolTelemetry={}
)
return Tool(
name="calculate",
description="Perform mathematical calculations. Supports basic arithmetic operations (+, -, *, /).",
parameters={
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "Mathematical expression to evaluate (e.g., '2 + 2 * 3')"
}
},
"required": ["expression"]
},
handler=calculate_handler
)
```
### 示例 3：随机数生成器
```python
import random
def create_random_number_tool():
"""创建随机数生成工具"""
async def random_handler(invocation: ToolInvocation) -> ToolResult:
try:
min_val = invocation["arguments"].get("min", 1)
max_val = invocation["arguments"].get("max", 100)
if min_val >= max_val:
raise ValueError("min must be less than max")
number = random.randint(min_val, max_val)
return ToolResult(
textResultForLlm=f"Generated random number: {number}",
resultType="success",
error=None,
toolTelemetry={}
)
except Exception as e:
return ToolResult(
textResultForLlm=f"Error: {str(e)}",
resultType="failure",
error=str(e),
toolTelemetry={}
)
return Tool(
name="generate_random_number",
description="Generate a random integer within a specified range.",
parameters={
"type": "object",
"properties": {
"min": {
"type": "integer",
"description": "Minimum value (inclusive)",
"default": 1
},
"max": {
"type": "integer",
"description": "Maximum value (inclusive)",
"default": 100
}
}
},
handler=random_handler
)
```
---
## 🔧 集成到 Pipe
### 完整的 Pipe 实现
```python
class Pipe:
class Valves(BaseModel):
# ... 现有 Valves ...
ENABLE_TOOLS: bool = Field(
default=False,
description="Enable custom tools (time, calculator, random)"
)
AVAILABLE_TOOLS: str = Field(
default="all",
description="Available tools: 'all' or comma-separated list (e.g., 'get_current_time,calculate')"
)
def __init__(self):
# ... 现有初始化 ...
self._custom_tools = []
def _initialize_custom_tools(self):
"""初始化自定义工具"""
if not self.valves.ENABLE_TOOLS:
return []
# 定义所有可用工具
all_tools = {
"get_current_time": create_time_tool(),
"calculate": create_calculator_tool(),
"generate_random_number": create_random_number_tool(),
}
# 根据配置过滤工具
if self.valves.AVAILABLE_TOOLS == "all":
return list(all_tools.values())
# 只启用指定的工具
enabled = [t.strip() for t in self.valves.AVAILABLE_TOOLS.split(",")]
return [all_tools[name] for name in enabled if name in all_tools]
async def pipe(
self,
body: dict,
__metadata__: Optional[dict] = None,
__event_emitter__=None,
__event_call__=None,
) -> Union[str, AsyncGenerator]:
# ... 现有代码 ...
# ✅ 初始化工具
custom_tools = self._initialize_custom_tools()
if custom_tools:
await self._emit_debug_log(
f"Enabled {len(custom_tools)} custom tools: {[t.name for t in custom_tools]}",
__event_call__
)
# ✅ 创建会话配置（传入工具）
from copilot.types import SessionConfig, InfiniteSessionConfig
session_config = SessionConfig(
session_id=chat_id if chat_id else None,
model=real_model_id,
streaming=body.get("stream", False),
tools=custom_tools, # ✅✅✅ 关键：传入工具列表
infinite_sessions=infinite_session_config if self.valves.INFINITE_SESSION else None,
)
session = await client.create_session(config=session_config)
# ... 其余代码保持不变 ...
```
---
## 📊 处理工具调用事件
### 在 stream_response 中显示工具调用
```python
async def stream_response(
self, client, session, send_payload, init_message: str = "", __event_call__=None
) -> AsyncGenerator:
# ... 现有代码 ...
def handler(event):
event_type = str(getattr(event.type, "value", event.type))
# ✅ 工具调用开始
if "tool_invocation_started" in event_type or "tool_call_started" in event_type:
tool_name = get_event_data(event, "tool_name", "")
if tool_name:
queue.put_nowait(f"\n\n🔧 **Calling tool**: `{tool_name}`\n")
# ✅ 工具调用完成
elif "tool_invocation_completed" in event_type or "tool_call_completed" in event_type:
tool_name = get_event_data(event, "tool_name", "")
result = get_event_data(event, "result", "")
if tool_name:
queue.put_nowait(f"\n✅ **Tool `{tool_name}` completed**\n")
# ✅ 工具调用失败
elif "tool_invocation_failed" in event_type or "tool_call_failed" in event_type:
tool_name = get_event_data(event, "tool_name", "")
error = get_event_data(event, "error", "")
if tool_name:
queue.put_nowait(f"\n❌ **Tool `{tool_name}` failed**: {error}\n")
# ... 其他事件处理 ...
# ... 其余代码 ...
```
---
## 🧪 测试示例
### 测试 1：询问时间
```
User: "What time is it now?"
Expected Flow:
1. Copilot 识别需要调用 get_current_time 工具
2. 调用工具（无参数或默认参数）
3. 工具返回: "Current time: 2026-01-26 15:30:00"
4. Copilot 回答: "The current time is 2026-01-26 15:30:00"
Pipe Output:
---
🔧 **Calling tool**: `get_current_time`
✅ **Tool `get_current_time` completed**
The current time is 2026-01-26 15:30:00
---
```
### 测试 2：数学计算
```
User: "Calculate 123 * 456"
Expected Flow:
1. Copilot 调用 calculate 工具
2. 参数: {"expression": "123 * 456"}
3. 工具返回: "The result of 123 * 456 is 56088"
4. Copilot 回答: "123 multiplied by 456 equals 56,088"
Pipe Output:
---
🔧 **Calling tool**: `calculate`
✅ **Tool `calculate` completed**
123 multiplied by 456 equals 56,088
---
```
### 测试 3：生成随机数
```
User: "Give me a random number between 1 and 10"
Expected Flow:
1. Copilot 调用 generate_random_number 工具
2. 参数: {"min": 1, "max": 10}
3. 工具返回: "Generated random number: 7"
4. Copilot 回答: "I generated a random number for you: 7"
```
---
## 🔍 调试技巧
### 1. 记录所有工具事件
```python
def handler(event):
event_type = str(getattr(event.type, "value", event.type))
# 记录所有包含 "tool" 的事件
if "tool" in event_type.lower():
event_data = {}
if hasattr(event, "data"):
try:
event_data = {
"type": event_type,
"data": str(event.data)[:200] # 截断长数据
}
except:
pass
self._emit_debug_log_sync(
f"Tool Event: {json.dumps(event_data)}",
__event_call__
)
```
### 2. 验证工具注册
```python
async def pipe(...):
# ...
custom_tools = self._initialize_custom_tools()
# 调试：打印工具信息
if self.valves.DEBUG:
tool_info = [
{
"name": t.name,
"description": t.description[:50],
"has_handler": t.handler is not None
}
for t in custom_tools
]
await self._emit_debug_log(
f"Registered tools: {json.dumps(tool_info, indent=2)}",
__event_call__
)
```
### 3. 测试工具处理函数
```python
# 单独测试工具
async def test_tool():
tool = create_time_tool()
# 模拟调用
invocation = {
"session_id": "test",
"tool_call_id": "test_call",
"tool_name": "get_current_time",
"arguments": {"format": "%H:%M:%S"}
}
result = await tool.handler(invocation)
print(f"Result: {result}")
```
---
## ⚠️ 注意事项
### 1. 工具描述的重要性
工具的 `description` 字段非常重要，它告诉 AI 何时应该使用这个工具：
```python
# ❌ 差的描述
description="Get time"
# ✅ 好的描述
description="Get the current date and time. Use this when the user asks 'what time is it', 'what's the date', or needs to know the current timestamp."
```
### 2. 参数定义
使用标准的 JSON Schema 定义参数：
```python
parameters={
"type": "object",
"properties": {
"param_name": {
"type": "string", # string, integer, boolean, array, object
"description": "Clear description",
"enum": ["option1", "option2"], # 可选：枚举值
"default": "default_value" # 可选：默认值
}
},
"required": ["param_name"] # 必需参数
}
```
### 3. 错误处理
总是捕获异常并返回有意义的错误：
```python
try:
result = do_something()
return ToolResult(
textResultForLlm=f"Success: {result}",
resultType="success",
error=None,
toolTelemetry={}
)
except Exception as e:
return ToolResult(
textResultForLlm=f"Error occurred: {str(e)}",
resultType="failure",
error=str(e), # 用于调试
toolTelemetry={}
)
```
### 4. 异步 vs 同步
工具处理函数可以是同步或异步：
```python
# 同步工具
def sync_handler(invocation):
result = calculate(invocation["arguments"])
return ToolResult(...)
# 异步工具（推荐）
async def async_handler(invocation):
result = await fetch_data(invocation["arguments"])
return ToolResult(...)
```
---
## 🚀 快速开始清单
- [ ] 1. 在 Valves 中添加 `ENABLE_TOOLS` 配置
- [ ] 2. 定义 2-3 个简单的工具函数
- [ ] 3. 实现 `_initialize_custom_tools()` 方法
- [ ] 4. 修改 `SessionConfig` 传入 `tools` 参数
- [ ] 5. 在 `stream_response` 中添加工具事件处理
- [ ] 6. 测试：询问时间、计算数学、生成随机数
- [ ] 7. 添加调试日志
- [ ] 8. 同步中文版本
---
## 📚 完整的工具事件列表
根据 SDK 源码，可能的工具相关事件：
- `tool_invocation_started` / `tool_call_started`
- `tool_invocation_completed` / `tool_call_completed`
- `tool_invocation_failed` / `tool_call_failed`
- `tool_parameter_validation_failed`
实际事件名称可能因 SDK 版本而异，建议先记录所有事件类型：
```python
def handler(event):
print(f"Event type: {event.type}")
```
---
**快速实现入口：** 从示例 1（获取时间）开始，这是最简单的工具，可以快速验证整个流程！
**作者：** Fu-Jie
**日期：** 2026-01-26