iFlow CLI Python SDK

参考：iFlow CLI Python SDK

iFlow CLI SDK 是用于与 iFlow CLI 进行编程交互的 SDK。它通过代理通信协议（ACP）允许开发者构建具有对话、工具执行和任务规划能力的 AI 驱动应用程序。

快速上手

安装

pip install iflow-cli-sdk

SDK 会自动检测并启动 iFlow 进程，无需手动配置。SDK 自动处理：

1. 检测 iFlow 是否已安装

2. 启动 iFlow 进程（如果未运行）

3. 查找可用端口并建立连接

4. 退出时自动清理资源

简单示例

import logging
# 将iflow_sdk模块的日志级别设置为WARNING或更高，避免打印信息过多
logging.getLogger('iflow_sdk').setLevel(logging.WARNING)
import asyncio
from iflow_sdk import IFlowClient, AssistantMessage, TaskFinishMessage

async def main():
    async with IFlowClient() as client:
        await client.send_message("你好，iFlow！")
        
        async for message in client.receive_messages():
            if isinstance(message, AssistantMessage):
                print(message.chunk.text, end="\n", flush=True)
            elif isinstance(message, TaskFinishMessage):
                break

await main()

Show code cell output

Hide code cell output

你好！有什么我可以帮你的吗？

简单查询

最简单的使用方式是通过 query 函数：

from iflow_sdk import query

response = await query("法国的首都是哪里？")
print(response) # 输出：法国的首都是巴黎。

法国的首都是巴黎。

核心概念

IFlowClient

IFlowClient 是与 iFlow CLI 交互的主要接口，管理 WebSocket 连接的生命周期：

from iflow_sdk import IFlowClient, IFlowOptions

# 使用默认配置（自动管理进程）
async with IFlowClient() as client:
    await client.send_message("你的问题")

    async for message in client.receive_messages():
        # 处理消息
        break

# 使用自定义配置
options = IFlowOptions(
    url="ws://localhost:8090/acp", # WebSocket URL
    auto_start_process=True, # 自动启动 iFlow
    timeout=30.0 # 超时时间（秒）
)
async with IFlowClient(options) as client:
    await client.send_message("你的问题")
    async for message in client.receive_messages():
        # 处理消息
        break

消息类型

SDK 支持多种消息类型，对应 iFlow 协议的不同响应：

AssistantMessage - AI 助手响应

AssistantMessage 包含 AgentInfo 支持，可以获取代理的详细信息：

import asyncio
from iflow_sdk import IFlowClient, AssistantMessage, TaskFinishMessage, AgentInfo

async def handle_assistant_message():
    async with IFlowClient() as client:
        await client.send_message("请介绍一下Python")
    
        async for message in client.receive_messages():
            if isinstance(message, AssistantMessage):
                print(message.chunk.text, end="\n", flush=True)
                # 访问代理信息（如果有）
                if message.agent_info:
                    print(f"代理ID: {message.agent_info.agent_id}")
                    if message.agent_info.task_id:
                        print(f"任务ID: {message.agent_info.task_id}")
                    if message.agent_info.agent_index is not None:
                        print(f"代理索引: {message.agent_info.agent_index}")
            elif isinstance(message, TaskFinishMessage):
                break

await handle_assistant_message()

Show code cell output

Hide code cell output

Python是一种高级编程语言，以其简洁易读的语法和强大的功能而闻名。它支持多种编程范式，包括面向对象、命令式、函数式和过程式编程。Python拥有丰富的标准库和第三方库，适用于Web开发、数据科学、人工智能、自动化脚本等多个领域。

以下是Python的一些关键特性：

1. **简单易学**：Python的语法设计强调代码的可读性，使得初学者能够快速上手。
2. **跨平台**：Python可以在多种操作系统上运行，如Windows、macOS、Linux等。
3. **解释型语言**：Python代码在运行时被逐行解释执行，无需预先编译成机器码。
4. **动态类型系统**：变量不需要显式声明类型，Python会在运行时自动推断。
5. **丰富的生态系统**：通过PyPI（Python Package Index）可以访问大量的开源库和工具。
6. **广泛应用**：
   - Web开发（如Django, Flask）
   - 数据分析与科学计算（如NumPy, Pandas, Matplotlib）
   - 机器学习与人工智能（如TensorFlow, PyTorch, Scikit-learn）
   - 自动化运维与脚本编写
   - 网络爬虫（如Scrapy, BeautifulSoup）

如果你有特定的问题或想深入了解某个方面，请告诉我！

AgentInfo 使用示例

from iflow_sdk import AgentInfo

# 从代理ID解析信息
agent_id = "subagent-task-abc123-2-1735123456789"
agent_info = AgentInfo.from_agent_id_only(agent_id)

if agent_info:
    print(f"代理ID: {agent_info.agent_id}")
    print(f"任务ID: {agent_info.task_id}")
    print(f"代理索引: {agent_info.agent_index}")
    print(f"时间戳: {agent_info.timestamp}")

代理ID: subagent-task-abc123-2-1735123456789
任务ID: task-abc123
代理索引: 2
时间戳: 1735123456789

ToolCallMessage - 工具调用

工具调用消息现在也包含 AgentInfo 信息和工具名称：

import asyncio
from iflow_sdk import IFlowClient, ToolCallMessage, ToolCallStatus, TaskFinishMessage

async def handle_tool_calls():
    async with IFlowClient() as client:
        # 注意：这个示例演示如何查看工具调用消息
        # iFlow 会要求提供完整路径和内容来创建文件
        await client.send_message("列出当前目录的文件")
        
        async for message in client.receive_messages():
            if isinstance(message, ToolCallMessage):
                print(f"状态: {message.status}")
                
                # 新增：工具名称
                if message.tool_name:
                    print(f"工具名称: {message.tool_name}")
                
                # 访问代理信息
                if message.agent_info:
                    print(f"代理ID: {message.agent_info.agent_id}")
                    
            elif isinstance(message, TaskFinishMessage):
                break

await handle_tool_calls()

状态: ToolCallStatus.IN_PROGRESS
工具名称: list_directory
状态: ToolCallStatus.COMPLETED
工具名称: list_directory

PlanMessage - 任务计划

import asyncio
from iflow_sdk import IFlowClient, PlanMessage, TaskFinishMessage

async def show_plan():
    async with IFlowClient() as client:
        await client.send_message("帮我创建一个Python项目结构")
        
        async for message in client.receive_messages():
            if isinstance(message, PlanMessage):
                print("执行计划：")
                for entry in message.entries:
                    status_icon = "✅" if entry.status == "completed" else "⏳"
                    print(f"{status_icon} [{entry.priority}] {entry.content}")
            elif isinstance(message, TaskFinishMessage):
                break

await show_plan()

Show code cell output

Hide code cell output

执行计划：
⏳ [low] 创建项目根目录
⏳ [low] 创建项目子目录（如src, tests, docs等）
⏳ [low] 创建必要的项目文件（如README.md, pyproject.toml等）
⏳ [low] 创建示例源代码文件
⏳ [low] 创建示例测试文件
执行计划：
✅ [low] 创建项目根目录
✅ [low] 创建项目子目录（如src, tests, docs等）
⏳ [low] 创建必要的项目文件（如README.md, pyproject.toml等）
⏳ [low] 创建示例源代码文件
⏳ [low] 创建示例测试文件
执行计划：
✅ [low] 创建项目根目录
✅ [low] 创建项目子目录（如src, tests, docs等）
✅ [low] 创建必要的项目文件（如README.md, pyproject.toml等）
⏳ [low] 创建示例源代码文件
⏳ [low] 创建示例测试文件
执行计划：
✅ [low] 创建项目根目录
✅ [low] 创建项目子目录（如src, tests, docs等）
✅ [low] 创建必要的项目文件（如README.md, pyproject.toml等）
✅ [low] 创建示例源代码文件
⏳ [low] 创建示例测试文件
执行计划：
✅ [low] 创建项目根目录
✅ [low] 创建项目子目录（如src, tests, docs等）
✅ [low] 创建必要的项目文件（如README.md, pyproject.toml等）
✅ [low] 创建示例源代码文件
✅ [low] 创建示例测试文件

TaskFinishMessage - 任务完成

import asyncio
from iflow_sdk import IFlowClient, AssistantMessage, TaskFinishMessage, StopReason

async def check_completion():
    async with IFlowClient() as client:
        await client.send_message("计算 1+1")
        
        async for message in client.receive_messages():
            if isinstance(message, AssistantMessage):
                print(message.chunk.text, end="", flush=True)
            elif isinstance(message, TaskFinishMessage):
                print()  # 换行
                if message.stop_reason == StopReason.END_TURN:
                    print("任务正常完成")
                elif message.stop_reason == StopReason.MAX_TOKENS:
                    print("达到最大令牌限制")
                break  # TaskFinishMessage 表示对话结束
await check_completion()

2
任务正常完成

常见用例

交互式聊天机器人

import asyncio
from iflow_sdk import IFlowClient, AssistantMessage, TaskFinishMessage

async def chatbot():
    print("iFlow 聊天机器人 (输入 'quit' 退出)")
    print("-" * 50)
    
    async with IFlowClient() as client:
        while True:
            user_input = input("\n你: ")
            
            if user_input.lower() in ['quit', 'exit', 'q']:
                print("再见！")
                break
            
            await client.send_message(user_input)
            
            print("iFlow: ", end="", flush=True)
            async for message in client.receive_messages():
                if isinstance(message, AssistantMessage):
                    print(message.chunk.text, end="", flush=True)
                elif isinstance(message, TaskFinishMessage):
                    print()  # 换行
                    break

await chatbot()

iFlow 聊天机器人 (输入 'quit' 退出)
--------------------------------------------------
iFlow: Hello! It seems like you've sent a message with just "qq". Could you please provide more context or let me know what you'd like help with? I'm here to assist you with any questions or tasks you might have.
再见！

流式响应处理

from iflow_sdk import query_stream
import asyncio

async def stream_example():
    prompt = "解释一下什么是升华现象"
    
    # query_stream 返回文本块的异步生成器
    async for chunk in query_stream(prompt):
        print(chunk, end="", flush=True)
    print()  # 最后换行

await stream_example()

Show code cell output

Hide code cell output

升华现象是指物质从固态直接转变为气态的过程，而不经过液态阶段。这一过程是物理变化，通常发生在特定的温度和压力条件下。

### 关键特点：
1. **直接转变**：固态物质直接变为气态，跳过液态。
2. **吸热过程**：升华需要吸收热量，称为升华热。
3. **可逆性**：气态物质也可以直接转变为固态，称为凝华。

### 常见例子：
- **干冰（固态二氧化碳）**：在常温下直接升华为气体，用于制冷和舞台效果。
- **樟脑丸**：在空气中逐渐升华，释放气味。
- **碘晶体**：加热时直接升华为紫色气体。

### 应用领域：
- **冷冻干燥**：用于保存食品和药品，通过升华去除水分。
- **化学实验**：分离和纯化某些固体物质。
- **日常应用**：如空气清新剂、防虫剂等。

升华现象在日常生活和工业中都有广泛应用，理解其原理有助于更好地利用这一特性。

高级配置

手动进程管理

如果需要手动管理 iFlow 进程：

import asyncio
from iflow_sdk import IFlowClient, IFlowOptions, AssistantMessage, TaskFinishMessage

async def manual_process_example():
    # 禁用自动进程管理
    options = IFlowOptions(
        auto_start_process=False,
        url="ws://localhost:8090/acp"  # 连接到已存在的 iFlow
    )
    
    async with IFlowClient(options) as client:
        await client.send_message("你的问题")
        async for message in client.receive_messages():
            if isinstance(message, AssistantMessage):
                print(message.chunk.text, end="", flush=True)
            elif isinstance(message, TaskFinishMessage):
                break

if __name__ == "__main__":
    asyncio.run(manual_process_example())

小技巧

注意 手动模式需要您单独启动 iFlow：

iflow --experimental-acp --port 8090

错误处理

SDK 提供了详细的错误处理机制：

import asyncio
from iflow_sdk import IFlowClient, ConnectionError, TimeoutError, AssistantMessage, TaskFinishMessage

async def error_handling_example():
    try:
        async with IFlowClient() as client:
            await client.send_message("测试")
            async for message in client.receive_messages():
                if isinstance(message, AssistantMessage):
                    print(message.chunk.text, end="", flush=True)
                elif isinstance(message, TaskFinishMessage):
                    break
    except ConnectionError as e:
        print(f"连接错误: {e}")
    except TimeoutError as e:
        print(f"超时错误: {e}")
    except Exception as e:
        print(f"未知错误: {e}")

await error_handling_example()

你好！看起来你可能想进行一些测试。你具体想测试什么呢？是关于代码、功能还是其他方面？请提供更多细节，我会尽力帮助你。

同步调用

对于需要同步调用的场景：

from iflow_sdk import query_sync, IFlowOptions

# 同步调用，带超时控制
options = IFlowOptions(timeout=30.0)
response = query_sync("你的问题", options=options)
print(response)

日志调试

启用详细日志以便调试：

import logging
from iflow_sdk import IFlowClient, IFlowOptions

# 设置日志级别
logging.basicConfig(level=logging.DEBUG)

options = IFlowOptions(log_level="DEBUG")
client = IFlowClient(options)