You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何通过Python代码调用Jupyter/IPython内核执行代码单元?工具选型咨询

工具选型指导

先明确各工具的定位,帮你理清选择方向:

  • jupyter_client:首选核心工具,属于Jupyter生态的底层客户端库,负责与任意Jupyter内核(Python、R、Rust等)建立连接、发送代码、接收执行结果,完全支持交互式状态保留,能覆盖你多语言执行、交互式调用的核心需求。
  • IPython:仅针对Python内核的交互库,若只处理Python代码可选用,但不支持多语言内核,灵活性远不如jupyter_client。
  • nbformat:(你提到的pyNotebook应该是指这个)用于解析Jupyter Notebook文件(.ipynb),可提取其中的Markdown和代码单元,配合jupyter_client处理混合格式的Notebook内容。
  • IRkernel/evcxr_jupyter等:这些是对应语言(R、Rust)的Jupyter内核实现,需安装后才能让jupyter_client执行对应语言的代码,属于内核服务而非客户端工具。
核心功能实现步骤

1. 环境准备

安装必要依赖:

pip install jupyter_client nbformat

按需安装对应语言的内核(比如R的IRkernel、Rust的evcxr_jupyter),按照官方文档完成配置。

2. 交互式执行代码单元

通过jupyter_client连接内核,按顺序发送代码即可保留执行状态:

import jupyter_client

# 指定目标内核(如python3、r、rust)
kernel_name = 'python3'
# 启动内核并创建客户端
km, kc = jupyter_client.manager.start_new_kernel(kernel_name=kernel_name)

def execute_code(code):
    # 发送代码执行请求
    msg_id = kc.execute(code)
    # 获取执行状态
    reply = kc.get_shell_msg(msg_id)
    status = reply['content']['status']
    output = []
    # 收集输出内容(标准输出/错误、执行结果)
    while True:
        try:
            msg = kc.get_iopub_msg(timeout=1)
            msg_type = msg['msg_type']
            content = msg['content']
            if msg_type == 'stream':
                output.append(f"{content['name']}: {content['text']}")
            elif msg_type == 'execute_result':
                output.append(f"Result: {content['data'].get('text/plain', '')}")
            elif msg_type == 'error':
                output.append(f"Error: {' '.join(content['traceback'])}")
        except jupyter_client.KernelTimeoutError:
            break
    return {
        'status': status,
        'output': '\n'.join(output)
    }

# 示例:先执行1-5单元,再执行第8单元
unit1_5 = """
a = 10
b = 20
c = a + b
print(c)
"""
unit8 = """
print(c * 2)
"""

# 执行1-5单元
result1 = execute_code(unit1_5)
print(result1['status'], result1['output'])

# 执行第8单元(内核保留了之前的变量c)
result2 = execute_code(unit8)
print(result2['status'], result2['output'])

# 关闭内核
km.shutdown_kernel()

3. 执行魔法命令

直接将魔法命令作为代码字符串发送即可,示例:

# 执行系统命令
result = execute_code("!pip list")
# 执行行魔法命令
result = execute_code("%cd /tmp")

4. 处理混合格式Notebook

用nbformat读取Notebook文件,提取并处理代码/Markdown单元:

import nbformat

# 读取Notebook文件
nb = nbformat.read('example.ipynb', as_version=4)
for cell in nb.cells:
    if cell.cell_type == 'code':
        # 执行代码单元
        result = execute_code(cell.source)
        print(f"Cell {cell.index} result: {result['status']}")
    elif cell.cell_type == 'markdown':
        # 自行处理Markdown内容(如提取文本)
        print(f"Markdown content: {cell.source}")

5. 实现干净执行环境

  • 每次启动新内核实例:调用start_new_kernel会创建独立内核,环境完全隔离。
  • 清空当前内核状态:执行%reset -f魔法命令快速清空变量。
  • 使用虚拟环境:创建独立Python虚拟环境,安装对应内核并指定运行路径,确保环境彻底隔离。
LLM函数调用适配

将执行逻辑封装为LLM可调用的标准函数,示例:

def run_jupyter_code(code: str, kernel: str = "python3") -> dict:
    """
    执行Jupyter代码单元并返回结果
    :param code: 要执行的代码字符串
    :param kernel: 内核名称(python3/r/rust等)
    :return: 包含执行状态(status)和输出(output)的字典
    """
    # 复用之前的execute_code逻辑,可根据需求选择复用内核或每次新建
    km, kc = jupyter_client.manager.start_new_kernel(kernel_name=kernel)
    result = execute_code(code)
    km.shutdown_kernel()
    return result

让LLM根据场景调用该函数,获取执行结果后继续交互流程。

VB.NET通过Python.NET调用IPython/Jupyter

1. 环境准备

  • 安装Python.NET:pip install pythonnet
  • 在VB.NET项目中添加对Python.Runtime.dll的引用(通常位于Python的site-packages/runtime目录下)。

2. VB.NET代码示例

Imports Python.Runtime

Module JupyterCaller
    Sub Main()
        ' 初始化Python环境
        PythonEngine.Initialize()
        
        Using Py.GIL()
            ' 导入jupyter_client模块
            Dim manager = Py.Import("jupyter_client.manager")
            ' 启动Python内核
            Dim kernelResult = manager.start_new_kernel(kernel_name:="python3")
            Dim km = kernelResult.Item(0)
            Dim kc = kernelResult.Item(1)
            
            ' 定义执行代码的逻辑
            Dim executeCode As Func(Of String, PyObject) = Function(code)
                                                               Dim msgId = kc.execute(code)
                                                               Dim reply = kc.get_shell_msg(msgId)
                                                               Dim status = reply.content.status.ToString()
                                                               Dim outputList = New List(Of String)()
                                                               Try
                                                                   Do
                                                                       Dim msg = kc.get_iopub_msg(timeout:=1)
                                                                       Dim msgType = msg.msg_type.ToString()
                                                                       Dim content = msg.content
                                                                       If msgType = "stream" Then
                                                                           outputList.Add($"{content.name.ToString()}: {content.text.ToString()}")
                                                                       ElseIf msgType = "execute_result" Then
                                                                           outputList.Add($"Result: {content.data("text/plain").ToString()}")
                                                                       ElseIf msgType = "error" Then
                                                                           outputList.Add($"Error: {String.Join(" ", content.traceback)}")
                                                                       End If
                                                                   Loop
                                                               Catch ex As PythonException
                                                                   ' 捕获超时异常,结束输出收集
                                                               End Try
                                                               Return PyDict.Create(New With {
                                                                   Key .status = status,
                                                                   Key .output = String.Join(vbNewLine, outputList)
                                                               })
                                                           End Function
            
            ' 执行测试代码
            Dim result = executeCode("print('Hello from VB.NET via Python.NET')")
            Console.WriteLine($"Status: {result("status")}")
            Console.WriteLine($"Output: {result("output")}")
            
            ' 关闭内核
            km.shutdown_kernel()
        End Using
        
        PythonEngine.Shutdown()
    End Sub
End Module

注意:VB.NET中处理Python对象时需注意类型转换,确保正确解析内核返回的字典、列表等结构。

内容的提问来源于stack exchange,提问作者Leroy Dyer

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.23 05:07:36